SkillClient 客户端

from tencentdb_agent_memory.v3 import SkillClient, AsyncSkillClient
from tencentdb_agent_memory.v3.skill_client import encode_utf8, encode_base64

from tencentdb_agent_memory.v3 import SkillClient

client = SkillClient(
    endpoint="https://memory.tdai.tencentyun.com",
    api_key="sk-xxxxxxxxxxxxxxxx",
    service_id="tdai-mem-xxxxxxxx",
    team_id="team-abc123",
    agent_id="agt-xyz789",
    user_id="usr-456",
    task_id="task-2026q3",
    timeout=30,
    verify=False,
)

resp = client.search(
    query="Python type hints usage",
    top_k=10,
    mode="hybrid",
)
for hit in resp["items"]:
    print(f"[{hit['score']:.2f}] {hit['name']}")

resp = client.list(
    filters={"team_id": "team-abc123"},
    pagination={"limit": 20, "offset": 0},
)
for skill in resp["items"]:
    print(f"{skill['name']} v{skill['version']}")

# 查询最新版本
resp = client.get("skill-abc123")
print(resp["name"], resp["content"])

# 查询指定版本
resp = client.get("skill-abc123", version=3)

# 仅查询元数据，不返回 content
resp = client.get("skill-abc123", include_content=False)

resp = client.versions(
    skill_id="skill-abc123",
    pagination={"limit": 10, "offset": 0},
)
for v in resp["items"]:
    print(f"v{v['version']}: {v['created_at']}")

resp = client.read_file(
    skill_id="skill-abc123",
    path="examples/usage.py",
    version=3,
)
print(resp["content"])

resp = client.listing(
    query="coding helper",
    char_budget=2000,
)
print(resp["block"])

# 构建资源文件
resource = client.encode_utf8(
    "README.md",
    "# Python Best Practices\\n\\n- Use type hints\\n- Follow PEP 8",
    mime_type="text/markdown",
)

# 创建 Skill
resp = client.create(
    name="python-tips",
    content="---\\nname: python-tips\\ndescription: Python coding tips\\n---\\n# Tips\\n",
    resources=[resource],
    metadata={"category": "programming"},
)
print(resp["skill_id"])  # "skill-abc123"

resp = client.extract(
    messages=[
        {"role": "user", "content": "请帮我写一个 Python 脚本来处理 CSV 文件"},
        {"role": "assistant", "content": "好的，以下是处理 CSV 的脚本..."},
    ],
    session_id="agent-main:sess-001",
)
print(resp["task_id"])  # 异步提取任务 ID

client.conversation_add(
    session_id="agent-main:sess-001",
    user_id="usr-456",
    team_id="team-abc123",
    agent_id="agt-xyz789",
    messages=[
        {"role": "user", "content": "帮我查一下部署配置"},
        {"role": "assistant", "content": "以下是当前服务的部署参数..."},
    ],
)

# 积累足够对话后，不传 messages，从缓冲区提取
resp = client.extract(session_id="agent-main:sess-001")

resp = client.update(
    skill_id="skill-abc123",
    expected_version=2,
    content="---\\nname: python-tips\\n---\\n# Updated Python Tips\\n",
)

resp = client.patch(
    skill_id="skill-abc123",
    expected_version=3,
    old_string="Use type hints",
    new_string="Use type hints (PEP 484)",
)

resp = client.write_files(
    skill_id="skill-abc123",
    expected_version=3,
    files=[
        client.encode_utf8("examples/usage.py", "print('hello')", mime_type="text/x-python"),
        client.encode_base64("assets/logo.png", b"...", mime_type="image/png"),
    ],
)

resp = client.remove_files(
    skill_id="skill-abc123",
    expected_version=4,
    paths=["examples/usage.py", "assets/logo.png"],
)

resp = client.delete(
    skill_id="skill-abc123",
    expected_version=5,
)

client = SkillClient(
    endpoint="https://memory.tdai.tencentyun.com",
    api_key="sk-xxx",
    service_id="tdai-mem-xxx",
    team_id="team-a",
    user_id="usr-one",
)

# 切换到不同团队和用户
client_b = client.with_defaults(team_id="team-b", user_id="usr-two")
result = client_b.search(query="Python best practices", top_k=5)

# 仅切换用户，团队沿用原始客户端的默认值
client_c = client.with_defaults(user_id="usr-three")

with SkillClient(
    endpoint="https://memory.tdai.tencentyun.com",
    api_key="sk-xxxxxxxxxxxxxxxx",
    service_id="tdai-mem-xxxxxxxx",
) as client:
    resp = client.search(query="Python tips", top_k=5)

import asyncio
from tencentdb_agent_memory.v3 import AsyncSkillClient

async def main():
    async with AsyncSkillClient(
        endpoint="https://memory.tdai.tencentyun.com",
        api_key="sk-...",
        service_id="tdai-mem-xxxxxxxx",
    ) as client:
        created = await client.create(
            name="python-tips",
            content="---\\nname: python-tips\\n---\\n# Python tips\\n",
        )
        print(f"创建成功: {created['skill_id']}")

        result = await client.search(query="Python tips", top_k=5)
        for hit in result["items"]:
            print(f"[{hit['score']:.2f}] {hit['name']}")

asyncio.run(main())