TxCodeClient API
TxCodeClient 是 txcode-sdk 的唯一入口类,封装了 Agent 管理、会话管理和上下文压缩。
导入
from txcode_sdk import TxCodeClient
构造函数
TxCodeClient(
api_key: str,
base_url: str = "https://api.openai.com/v1",
model: str = "gpt-4",
agent_type: AgentType = "code",
tools: list[Tool] | None = None,
max_context_tokens: int = 150_000,
max_iterations: int = 50,
system_prompt: str | None = None,
work_dir: str | None = None,
log_enabled: bool = False,
)
详见 配置指南。
chat()
发送消息给 Agent,执行 ReAct 循环。
async def chat(
self,
message: str,
session_id: str | None = None,
agent_type: AgentType | None = None,
history: list[dict] | None = None,
media_files: list[MediaFile] | None = None,
on_step: Callable[[StepInfo], None] | None = None,
on_compact: Callable[[dict], None] | None = None,
abort_signal: Any = None,
) -> ChatResult:
返回值 ChatResult:
| 字段 | 类型 | 说明 |
|---|---|---|
answer | str | Agent 最终回答 |
steps | list[StepInfo] | 所有工具调用步骤 |
session_id | str | 会话 ID |
iterations | int | 实际迭代次数 |
success | bool | 是否成功 |
error | str | None | 错误信息 |
usage | TokenUsage | None | Token 用量 |
续接会话
result1 = await client.chat("我叫小明")
sid = result1.session_id
result2 = await client.chat("我叫什么名字?", session_id=sid)
print(result2.answer) # "小明"
on_step 回调
from txcode_sdk.types import StepInfo
def on_step(step: StepInfo):
print(f"第 {step.iteration} 步")
for i, tc in enumerate(step.tool_calls):
res = step.results[i] if i < len(step.results) else None
status = "✓" if res and res.success else "✗"
print(f" [{status}] {tc.name}: {tc.arguments}")
result = await client.chat("列出所有 .py 文件", on_step=on_step)
abort_signal
class AbortSignal:
aborted = False
signal = AbortSignal()
task = asyncio.create_task(client.chat("复杂任务", abort_signal=signal))
会话管理
list_sessions()
for s in client.list_sessions():
print(s["print_summary"])
get_session()
session = client.get_session(session_id)
get_session_messages()
data = client.get_session_messages(session_id)
for msg in data["messages"]:
print(msg.get("printMsg") or msg["content"][:200])
delete_session()
client.delete_session(session_id)
完整示例
import asyncio
from txcode_sdk import TxCodeClient
from txcode_sdk.types import StepInfo
async def main():
client = TxCodeClient(
api_key="sk-xxx",
base_url="https://api.deepseek.com/v1",
model="deepseek-chat",
max_iterations=10,
)
def on_step(step: StepInfo):
if step.printMsg: print(step.printMsg)
result = await client.chat("在 src/ 下找到所有 Python 文件", on_step=on_step)
print(result.answer)
asyncio.run(main())