Metadata-Version: 2.4
Name: yaoagent
Version: 0.1.0
Summary: 声明式、响应式的 Python 智能体框架，灵感来自 Apple Foundation Models 的 dynamic sessions API。
License: MIT
Project-URL: Documentation, https://github.com/HawkonLi/yao_agent/blob/main/docs/GUIDE.md
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: openai>=1.0
Requires-Dist: pydantic>=2.0
Requires-Dist: pyyaml>=6.0
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"
Dynamic: license-file

# YaoAgent

>声明你的智能体结构和编排，简化在科研或其他轻量场景下智能体框架带来的额外负担，让智能体编排像声明界面一样清晰和简单

YaoAgent提供了Instruction、Profile、会话管理和编排工具以及日志和追踪工具，用于声明式定义单智能体和多智能体任务。框架使用生命周期管理能够
在编排时关注请求、智能体、工具等各个模块，使用`EnvironmentObject`进行跨会话数据流管理，使用修饰符允许快速配置各种参数信息。借助成熟的声明式
UI的范式，让你能够只聚焦在智能体的编排中。**如果你对声明式UI不熟悉，可以从这里开始**：[教学指南 docs/GUIDE.md](docs/GUIDE.md)

Inspired By Apple [Foundation Models(iOS 27+)](https://developer.apple.com/documentation/foundationmodels/composing-dynamic-sessions-with-instructions-and-profiles) and [SwiftUI](https://developer.apple.com/documentation/SwiftUI)



## 安装

```bash
pip install openai pydantic pyyaml
```

在项目目录放一个 `.env`（框架会自动向上查找加载）：

```
DEEPSEEK_API_KEY=sk-...
```

## 快速上手

```python
import asyncio
from typing import Annotated
from yaoagent import *

class GetWeather(Tool):
    name: str = "get_weather"
    description: str = "查询城市天气。"
    def call(self, city: Annotated[str, "城市名"]) -> str:   # 参数 schema 自动生成
        return f'{{"city": "{city}", "temp": 22}}'

class Assistant(DynamicInstructions):
    def body(self, session) -> DynamicInstructionStream:     # 生成器 = 声明式组合
        yield Instructions("你是天气助手，需要时调用工具。")
        yield GetWeather()
        if getattr(session.state, "verbose", False):         # 随会话状态响应式分支
            yield Instructions("回答尽量详细。")

async def main():
    session = LanguageModelSession(
        Assistant(),
        llm_config=LLMConfig.deepseek("deepseek-v4-flash"),
    )
    print(await session.respond("北京天气怎么样？"))

asyncio.run(main())
```

## 核心概念

| 类型 | 作用 |
|---|---|
| `Instructions` | 一段模型可见的指令文本。 |
| `Tool` | 可被模型调用的能力；`call` 的类型注解自动生成参数 schema，支持同步/异步。 |
| `DynamicInstructions` | `body()` 是生成器，用 `yield` 声明指令、工具、嵌套指令；每次请求前重新求值。 |
| `Profile` | 绑定一组动态指令 + 模型参数（model/temperature/reasoning）+ 生命周期钩子，不可变。 |
| `DynamicProfile` | `body()` 按状态选出唯一一个激活 `Profile`，用于编排多个领域子配置。 |
| `LanguageModelSession` | 一个智能体：持有配置、私有状态 `state`、共享环境、历史；`respond()` / `stream_response()` 发起请求。 |
| `EnvironmentObject` / `Environment` | 跨智能体共享的对象，按类型注入与取用（≈ SwiftUI `@EnvironmentObject`）。 |
| `SessionGroup` | 把多个智能体按拓扑（串/并/循环）编排成一个可嵌套、可 `run` 的整体。 |

三层结构：`DynamicProfile`（选哪个） → `Profile`（参数/钩子） → `DynamicInstructions`（指令/工具）。

## 链式修饰符（命名对齐 Swift DSL）

修饰符与 Swift 同名，链式书写；既能挂在终态 `Profile`，也能挂在外层 `DynamicProfile` 并“穿透”到内层：

```python
class MyProfile(DynamicProfile):
    def body(self, session) -> Profile:
        return (Profile(instructions=MyInstructions())
                .model("deepseek-v4-pro")
                .temperature(0.7)
                .reasoning("high")
                .on_tool_call(lambda c: log(c))      # 生命周期钩子
                .history_transform(lambda h: h[-20:]))
```

- **值类**（`.model/.temperature/.reasoning/.history_transform`）：外层作默认值，**内层优先**。
- **钩子类**（`.on_*`）：跨层**累加**，外层先触发。

### 可复用的自定义修饰符

把一组成套的参数与钩子封装成一个可命名、可复用、可链式挂载的修饰符
（对应 Apple 的 `DynamicProfileModifier` / `.modifier(_:)`）。可以是函数，也可以是类：

```python
# 函数形式：返回 Profile -> Profile
def staged(label: str) -> ProfileModify:
    return lambda p: (p.on_activate(lambda: print(f">> {label}"))
                       .on_deactivate(lambda: print(f"<< {label}")))

# 类形式：实现 body(content)
class Debug(DynamicProfileModifier):
    def body(self, content: Profile) -> Profile:
        return content.temperature(0.0).on_response(lambda r: print(r))

Profile(instructions=MyInstructions()).temperature(0.8).modifier(staged("写作"))
Profile(instructions=MyInstructions()).modifier(Debug())
```

## 参数优先级（从高到低）

1. 调用点：`respond(prompt, temperature=0.0)`
2. `with` 临时重写：`with session.using(temperature=0.0): ...`（块内生效，离开还原）
3. 配置层：`Profile` / `DynamicProfile` 上的取值

## 生命周期钩子

链式声明，可同步或异步；钩子可闭包捕获 `session` 以读写会话状态。

| 钩子 | 触发时机 |
|---|---|
| `on_prompt(fn)` | 发起请求前（入参 prompt） |
| `on_response(fn)` | 得到最终回复后（入参 text；可在此压缩历史） |
| `on_tool_call(fn)` | 执行工具前（入参 `ToolCall`；**抛异常即拒绝**） |
| `on_tool_output(fn)` | 工具产出后（入参 `ToolCall, output`） |
| `on_activate(fn)` | 配置成为激活态时（适合初始化） |
| `on_deactivate(fn)` | 配置被切换走时（适合清理） |

`on_activate`/`on_deactivate` 由顶层 `DynamicProfile` 切换激活子配置时自动触发。

## 工具访问会话状态

工具默认是隔离的。需要会话时，用 `self.session` 访问即可（对标 Apple 的 `@SessionProperty`）——
`call` 签名保持纯净，只放模型参数；框架在工具执行期间自动绑定当前会话。借此读写
`session.state` / `session.history`，实现有状态工具（记忆、技能激活、给工具传上下文等）：

```python
class RememberTool(Tool):
    name: str = "remember"
    description: str = "记住一项用户偏好。"
    def call(self, key: str, value: str) -> str:   # 签名纯净，不掺框架参数
        self.session.state.prefs[key] = value      # self.session 自动可用
        return f"已记住 {key}={value}"

# 用 prefs={} 初始化会话状态，工具体里就无需处理默认值
session = LanguageModelSession(MyProfile(), llm_config=cfg, prefs={})
```

## 私有状态与共享环境

两层状态，边界清楚：

- **私有 `state`（≈ `@State`）**：会话自己拥有、跨请求持久。推荐传入**显式类型化对象**（dataclass），
  比无类型口袋安全：
  ```python
  @dataclass
  class KitchenState:
      stage: str = "discover"
      cart: list[str] = field(default_factory=list)

  session = LanguageModelSession(KitchenProfile(), llm_config=cfg, state=KitchenState())
  # body / 工具里 session.state.stage —— 类型已知、IDE/mypy 可查
  # （不传 state 时，关键字参数仍会汇成一个 SimpleNamespace，方便快速脚本）
  ```

- **共享 `environment`（≈ `@EnvironmentObject`）**：跨智能体共享的对象，**按类型**注入与取用：
  ```python
  class Notebook(EnvironmentObject):
      def __init__(self): self.findings = []

  class SaveFinding(Tool):
      name: str = "save_finding"; description: str = "记一条发现"
      notebook = Environment(Notebook)               # 按类型注入，不进 schema
      def call(self, text: str) -> str:
          self.notebook.findings.append(text); return "已记录"

  session.environment(Notebook())                    # 链式注入，可多个（一个类型一个实例）
  ```
  并发下保持「单一写者 + 同步读快照」即安全；需要跨 `await` 的多步更新就在你的环境对象里放一把
  `asyncio.Lock`。

## 多智能体编排

把多个会话（每个是一个智能体）按拓扑组合成一个可嵌套、可 `run` 的 `SessionGroup`：

```python
pipeline = (
    SessionGroup(
        parallel(researcher_a, researcher_b),          # 并行：同输入扇出
        synthesizer,                                   # 串行：上一步输出喂下一步
        loop(reviser, until=lambda o: "[OK]" in o, max_iters=3),  # 迭代到满足条件
    )
    .group_style(Style.sequential)                     # 顶层用串行把三段连起来
    .environment(Notebook())                           # 环境向所有成员（含嵌套子组）穿透
)
answer = await pipeline.run("研究主题")
```

group 由三个**正交维度**描述（编排约束另外两个的合法取值）：

- **编排 `group_style`**（成员怎么跑）：`Style.sequential` / `Style.parallel` / `Style.loop(until=, max_iters=)`。
- **输入 `input_style`**（成员收什么）：`InputStyle.pipe`（上一个输出喂下一个）/ `InputStyle.broadcast`（都拿原输入，靠共享 `environment` 通信）。
- **输出 `output_style`**（谁的输出暴露给 group）：`OutputStyle.last` / `OutputStyle.pick(member)` / `OutputStyle.merge(fn)`。

（`input_style` / `output_style` 命名描述的是**智能体之间的内部接线**，与面向用户的运行时输出层
`Runtime`（见下文）刻意区分。）每种编排自带默认输入/输出（如 sequential = pipe + last，
parallel = broadcast + merge），按需覆盖；非法组合（如 parallel + pipe）会报错。
便捷构造 `sequential() / parallel() / loop()` 即"编排 + 默认输入输出"。

```python
# 顺序跑、但成员各拿原输入、靠共享环境通信、返回末位成员（而非管道）：
SessionGroup(a, b).group_style(Style.sequential).input_style(InputStyle.broadcast).output_style(OutputStyle.last)
```

- 成员可以是会话，也可以是另一个 `SessionGroup`——递归嵌套。
- 并行就是 `asyncio.gather`；通信走共享 `environment`（黑板）或上下游的数据流。

## 会话历史与续接

`session.history` 是 OpenAI 格式的完整 transcript：包含 user 提示、工具调用、工具输出与
最终回复（不含指令）。可用既有历史种子初始化以续接对话：

```python
session = LanguageModelSession(MyProfile(), llm_config=cfg, history=prior_messages)
```

历史可在 `on_response` 钩子里压缩，或用 `.history_transform()` 在请求前做局部裁剪。

## 流式输出

`stream_response()` 是 `respond()` 的流式版：异步逐段产出最终回复的文本增量，
工具调用循环在内部静默处理，流结束后照常持久化完整 transcript。

```python
async for delta in session.stream_response("北京天气怎么样？"):
    print(delta, end="", flush=True)
```

底层模型把**答案**（content）和**思考**（reasoning，DeepSeek 推理模型）放在同一条流的不同字段里，
框架把它们 demux 成两个独立钩子，按需各接各的、零分支：

```python
Profile(instructions=...)
    .on_response_stream(handle_answer)     # 答案增量
    .on_reasoning_stream(handle_thinking)  # 思考增量（不想要就不写这行）
```

`stream_response()` 产出的流只含答案；思考只走 `on_reasoning_stream`，不混进答案、不进 transcript。

## 日志与可观测（实验复现）

绑一个 `Trace`，框架就在每个关键节点发**结构化事件**；`sink` 就是个 `Callable[[dict], None]`：

```python
session = LanguageModelSession(
    profile, llm_config=cfg,
    trace=Trace(jsonl("runs/exp1.jsonl"), console, level="debug"),
)
```

- 事件类型：`request`（含解析后的**完整配置快照**）/ `tool_call` / `tool_output` /
  `response`（含 token 用量与 `elapsed_ms`）/ `activate` / `deactivate` / `error`；
  多智能体编排另有 `group_start` / `group_end` / `member_start` / `member_end` / `iteration`。`debug` 级近乎全量。
- **关联 ID**：同一次 run（含其编排里所有成员/工具轮次）的事件共享一个 `run_id`，并发/嵌套时可归并到一条时间线。
- `SessionGroup.trace(t)` 把日志向所有成员穿透（成员自带的优先），整组事件自动带同一个 `run_id`。
- 内置 sink：`jsonl(path)`（一行一条，适合实验）、`console`；自定义就传任意 `lambda e: ...`。
- 接 SwanLab 等外部实验平台：`Trace(lambda e: swanlab.log(e))`——**适配器写在你的实验代码里，不进框架**。
- `session.describe()` 可随时导出当前解析出的配置快照（指令 / 工具 schema / 模型参数 / 状态）。

## 运行时输出封装（Runtime / Handler）

把“输出往哪送”从“生命周期里发生了什么”里拆出来，统一成 **Handler**——一组**形状 = 钩子**的方法
（`tool_call` / `response` / `response_stream` …），每个把事件打成 dict 丢给同一个 `sink`（目的地）。
`Runtime` 是装三个 Handler 的盒子，按受众分三个投递口：

| 投递口 | 给谁 | 典型去处 |
|---|---|---|
| `log` | 开发者 / 留档 | Trace → jsonl / console |
| `stream` | 最终用户（实时） | SSE / websocket / 终端 |
| `output` | 上游系统 | 结构化 JSON |

`Runtime` **永远有默认值**（模块级默认 + `ContextVar`），`session.runtime` 任何时候都拿得到非空对象。
在 `body` 里取投递口、把想要的事件接上去即可（不想要就不接，零分支）：

```python
class MyProfile(DynamicProfile):
    def body(self, session) -> Profile:
        io = session.runtime
        return (
            Profile(instructions=MyInstructions())
            .on_response_stream(io.stream.response_stream)  # 答案增量 → 实时给用户
            .on_tool_output(io.log.tool_output)             # 工具输出 → 留档
            .on_response(io.output.response)                # 最终回复 → 结构化给上游
        )
```

## App 级封装（部署 / 集成边界）

`Session` / `SessionGroup` 是“View”（可组合的智能体逻辑）；`App` 是把它们接到外部世界
（FastAPI、推荐系统、命令行）的最外层外壳——**可选**，框架内直接 `respond()` 即可。模板方法：
框架定 `run()` 骨架，你只实现 `body()`，按需覆写对外通道。

```python
class ResearchApp(App):
    def body(self):                       # 要跑什么（每次 run 新建一份 → 请求间隔离）
        return SessionGroup(...).llm_config(cfg)
    def on_stream(self, event): ...       # 运行中：流式增量往哪送（默认 no-op）
    def on_log(self, event): ...          # 运行中：日志往哪送（同时收进信封 events）

envelope = await ResearchApp().run("电动汽车的未来")
# {run_id, output, usage, finish_reason, elapsed_ms, events}

# 想要别的返回形状：覆写 run 调 super() 拿信封再加工（标准 Python，复用全部样板）
class RecApp(App):
    def body(self): return SessionGroup(...).llm_config(cfg)
    async def run(self, input):
        env = await super().run(input)
        return {"items": parse(env["output"]), "cost": env["usage"]}
```

- **隔离**：每次 `run()` 由 `body()` 新建一份 Runnable，并在自己的 `Runtime` + `run_scope` 里执行（基于 ContextVar），并发互不串。
- **统一出口**：`run()` 返回标准 JSON 信封，便于对接推荐系统等下游。
- ResearchApp / RecApp 各继承 `App`：运行中通道用 `on_*` 覆写，最终形状用覆写 `run` 调 `super()`——骨架不动。
- `SessionGroup` 与会话一样统一返回 `Response`：文本由 `output_style` 决定，`usage` 是
  **全编排所有成员（含嵌套子组、loop 各轮）的累加**（即这次多智能体跑的总成本），故信封 `usage` 对 group 也正常。

### 两种交付面：`run()` 批量 / `stream()` 实时

同一个 `body()`（模型层），两种交付：

- **`run()`** → 阻塞、返回结构化 JSON 信封。适合**离线实验 / 打分 / 后端**（"研究面"）。
- **`stream()`** → 异步产出**标准 UI 事件流**，给真实**对话助手**前端实时渲染（"服务面"）。

```python
async for event in MyApp().stream("上海今天穿什么？先查天气"):
    # event["type"] ∈ {text, reasoning, tool_call, tool_output, progress, done, error}
    render(event)
```

事件含：`text`（答案增量）/ `reasoning`（思考增量）/ `tool_call` / `tool_output` /
`progress`（进展流程：group/member/iteration/activate）/ `done`（最终结果）/ `error`。
`body()` 为单会话时逐 token 流式产出 `text`；为 group 时产出进展/工具事件、最终文本随 `done` 给出。
内部用 asyncio 队列把运行中各处事件汇成一条可 `async for` 的流（见 [example_app.py](example_app.py) 场景 5）。

## 输入（Prompt）

`respond()` 的入参就是 `str`，绝大多数场景直接传字符串即可。需要携带元数据/附件时用 `Prompt`
（`str` 子类，与 `Response` 对称，零破坏）：

```python
await session.respond(Prompt("分析这段", metadata={"lang": "zh"}))   # 钩子里可读 prompt.metadata
```

`attachments` 字段预留给将来的多模态输入。

## 返回值与用量

`respond()` 返回 `Response`——它是 `str` 子类（可直接打印/比较/拼接），额外携带
token 用量与结束原因：

```python
answer = await session.respond("北京天气怎么样？")
print(answer)              # 回复文本
print(answer.usage)        # Usage(prompt_tokens=..., completion_tokens=..., total_tokens=...)（跨工具轮次累加）
print(answer.finish_reason)
```

## 错误系统

所有框架错误都是 `YaoError` 子类，带稳定错误码与自然语言解释：

```python
try:
    await session.respond("...")
except ToolError as e:       # ConfigError / ResolveError / ToolError / ModelError
    e.code           # ErrorCode.INVALID_TOOL_ARGUMENTS
    str(e)           # "[YAO-3002] 工具参数不符合其 schema。 (tool='...')"
    e.explain()      # 面向人/模型的自然语言解释
    e.to_dict()      # 结构化暴露：code/name/explanation/context/cause
```

**参数校验自愈**：模型把工具叫错或参数不合法（可恢复错误）时，框架不会中断会话，
而是把自然语言解释作为工具结果回灌给模型，让它在下一轮自行改正；工具自身执行失败
（致命错误）才向上抛出。

## 在 Web 服务里用（FastAPI/Django）

框架是纯 asyncio，和 FastAPI 异步端点天然契合；`AsyncOpenAI` 客户端按事件循环 + 连接参数自动复用，
轻量并发没问题。几条纪律：

- **一对话一 session**：别把同一个可变 session 跨并发请求共享（`history`/`state` 会被写乱）。
- **共享 environment 按用户域**：全局可变就用单一写者 + 锁；注意多 worker 是多进程，
  内存对象不跨进程——要横向扩展就把共享状态外置到 Redis/DB。
- **工具别阻塞事件循环**：阻塞 IO 用 `async def call` 或 `asyncio.to_thread`。

## 运行示例

```bash
python3 example.py          # 能力速览 + 多阶段厨房编排智能体
python3 example_group.py    # 完整多智能体 DSL：并行调研 → 综述 → 自我精修，共享笔记本环境
```

- `example.py`：① 能力速览（响应式指令、自动 schema、生命周期钩子、`with` 重写 + reasoning、
  穿透传值、参数校验、错误系统、流式）；② 多阶段厨房助手（顶层 `DynamicProfile` 按阶段切换子配置）。
- `example_group.py`：用 `SessionGroup` 把多个智能体编排成 `sequential( parallel(...) → 综述 → loop(...) )`，
  并通过共享 `Notebook` 环境协作——集中体现多智能体 + 环境 + 嵌套拓扑。

## 许可证

[MIT](LICENSE)。
