# piagentcore **Repository Path**: cc11e/piagentcore ## Basic Information - **Project Name**: piagentcore - **Description**: No description available - **Primary Language**: Unknown - **License**: MIT - **Default Branch**: main - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-08-29 - **Last Updated**: 2026-09-01 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # pi-agent-core 一个面向 Python 的有状态 LLM 智能体框架，提供工具执行、事件流、转向/后续消息队列以及代理传输。 ## 安装 ```bash uv add pi-agent-core ``` 可选的 Anthropic 适配器： ```bash uv add "pi-agent-core[anthropic]" ``` ## 概述 pi-agent-core 提供了一个极简、与 LLM 无关的智能体循环，负责在你的应用与任意 LLM 提供商之间进行编排。你只需提供自己的流式函数——本库负责状态管理、工具执行、事件分发、回合中转向（steering）以及后续队列。 ### 主要特性 - **与 LLM 无关** —— 通过 `StreamFn` 兼容任意提供商 - **实时事件流** —— 两级事件系统，覆盖智能体生命周期与 LLM 流式原语 - **工具执行** —— 使用 JSON Schema 参数与异步执行函数来定义工具 - **转向与后续队列** —— 可在回合中打断，或在完成后排队消息 - **取消** —— 通过 `asyncio.Event` 实现协作式取消 - **代理传输** —— 内置 SSE 代理客户端，可将请求路由到后端服务器 - **完整类型标注** —— 全程使用 Pydantic 模型，并带有 `py.typed` 标记 ## 快速开始 ```python import asyncio from pi_agent_core import ( Agent, AgentOptions, AgentEvent, AgentTool, AgentToolSchema, AgentToolResult, Model, TextContent, ) # 1. Define your tools async def greet(tool_call_id, params, cancel_event=None, on_update=None): name = params.get("name", "world") return AgentToolResult(content=[TextContent(text=f"Hello, {name}!")]) greet_tool = AgentTool( name="greet", description="Greet someone by name", parameters=AgentToolSchema( properties={"name": {"type": "string", "description": "Name to greet"}}, required=["name"], ), execute=greet, ) # 2. Implement a StreamFn for your LLM provider # (see "Implementing a StreamFn" section below) async def my_stream_fn(model, context, options): ... # 3. Create and run the agent agent = Agent(AgentOptions(stream_fn=my_stream_fn)) agent.set_model(Model(api="anthropic", provider="anthropic", id="claude-sonnet-4-20250514")) agent.set_system_prompt("You are a helpful assistant.") agent.set_tools([greet_tool]) # Subscribe to events def on_event(event: AgentEvent): print(f"Event: {event.type}") agent.subscribe(on_event) # Send a prompt asyncio.run(agent.prompt("Say hello to Alice")) ``` ## 架构 ``` Agent ← 高层有状态封装、订阅、队列 ↓ agent_loop() ← 核心编排：prompt → stream → tools → steering loop ↓ StreamFn (user-provided) ← 由你实现 LLM 流式集成 or stream_proxy() ← 内置 SSE 代理客户端，作为 StreamFn 使用 ``` ### 模块 | 模块 | 职责 | |---|---| | `types.py` | 所有 Pydantic 模型：内容块、消息、事件、工具、配置、状态，以及 `StreamResult` | | `agent_loop.py` | `agent_loop()` 与 `agent_loop_continue()` 异步生成器——流式处理、工具执行、转向、后续处理 | | `agent.py` | `Agent` 类，封装循环并负责状态管理、事件订阅、中止/重置以及队列管理 | | `proxy.py` | `stream_proxy()` SSE 客户端，基于 httpx——从服务器剥离的增量事件中重建部分消息 | | `anthropic.py` | 可选的 Anthropic Messages 适配器（`stream_anthropic`） | ## 实现 StreamFn 本库与 LLM 无关。你需提供一个 `stream_fn(model, context, options)`，返回一个**过程式**的 `StreamResult` 字典： - `events`: `AsyncIterator[AssistantMessageEvent]` - `result`: `Callable[[], Awaitable[AssistantMessage]]` ```python from collections.abc import AsyncIterator from pi_agent_core import ( AssistantMessage, AssistantMessageEvent, StreamResult, StreamStartEvent, StreamTextStartEvent, StreamTextDeltaEvent, StreamTextEndEvent, StreamDoneEvent, TextContent, Model, AgentContext, SimpleStreamOptions, ) async def my_stream_fn( model: Model, context: AgentContext, options: SimpleStreamOptions, ) -> StreamResult: partial = AssistantMessage(api=model.api, provider=model.provider, model=model.id) async def events_iter() -> AsyncIterator[AssistantMessageEvent]: # Example only — replace with provider events yield StreamStartEvent(partial=partial) partial.content = [TextContent(text="")] yield StreamTextStartEvent(content_index=0, partial=partial) partial.content[0].text += "Hello!" yield StreamTextDeltaEvent(content_index=0, delta="Hello!", partial=partial) yield StreamTextEndEvent(content_index=0, content="Hello!", partial=partial) partial.stop_reason = "stop" yield StreamDoneEvent(reason="stop", message=partial) async def result() -> AssistantMessage: return partial return {"events": events_iter(), "result": result} ``` ## 事件系统 ### 智能体事件（10 种） 覆盖智能体生命周期、回合、消息以及工具执行： `agent_start`、`agent_end`、`turn_start`、`turn_end`、`message_start`、`message_update`、`message_end`、`tool_execution_start`、`tool_execution_update`、`tool_execution_end` ### 助手消息事件（12 种） 覆盖由循环内部消费的 LLM 流式原语： `start`、`text_start`、`text_delta`、`text_end`、`thinking_start`、`thinking_delta`、`thinking_end`、`toolcall_start`、`toolcall_delta`、`toolcall_end`、`done`、`error` ## 转向与后续队列 转向消息会在回合中打断智能体（跳过剩余的工具调用）： ```python agent.steer(UserMessage(content=[TextContent(text="Actually, use a different approach")])) ``` 后续消息会在当前运行完成后触发新的回合： ```python agent.follow_up(UserMessage(content=[TextContent(text="Now summarize the results")])) ``` 两者均支持 `"one-at-a-time"`（默认）或 `"all"` 出队模式。 ## 内置适配器 ### Anthropic ```python from pi_agent_core.anthropic import stream_anthropic agent = Agent(AgentOptions( stream_fn=stream_anthropic, )) ``` 会自动使用 `OPENROUTER_API_KEY` 或 `ANTHROPIC_API_KEY`。你也可以设置 `ANTHROPIC_BASE_URL`。 ### 代理传输 适用于通过后端服务器路由 LLM 调用的应用： ```python from pi_agent_core import Agent, AgentOptions, stream_proxy, ProxyStreamOptions agent = Agent(AgentOptions( stream_fn=lambda model, context, options: stream_proxy( model, context, ProxyStreamOptions( **options.model_dump(), auth_token="your-auth-token", proxy_url="https://your-proxy.example.com", ), ), )) ``` ## 文档 中文学习与设计文档入口：[docs/README.md](docs/README.md) | 分区 | 内容 | |------|------| | 学习 | 路线总览、运行时序、`agent.py` 教学 | | 机制分册 | 01→07 对照源码精读 | | 工具治理 | Demo 层 Registry / 动态加载 | | 设计规格 | Permission、RBAC、Session、History | 旁路 Demo 启动：[apps/README.md](apps/README.md) ## 开发 ```bash uv sync # Install dependencies uv run pytest # Run all tests uv run pytest -v --tb=short # Verbose with short tracebacks uv run ruff check . # Lint uv run ruff format . # Format ``` ## 致谢 这是来自 **pi-mono** 仓库的 TypeScript 包 [`@mariozechner/pi-agent-core`](https://github.com/badlogic/pi-mono) 的 Python 移植版。原始的 TypeScript 实现由 [Mario Zechner](https://github.com/mariozechner) 提供，本库忠实地复刻了其架构、抽象与设计。 ## 许可证 [MIT](LICENSE)