两个主抽象:Agent 与 Task
30 秒导读: Upsonic 里你几乎永远只碰两个对象——
Task(一次「要做的事」:问题描述、能用的工具、参考上下文、期望的返回格式)和Agent(一个「会做事的人」:挂着模型、记忆、安全护栏、可靠性层)。写好这两个,调一句agent.do(task),框架 就替你把中间那条又长又脏的管线跑完。本章讲清这两个对象各装了什么、do一族入口分别在什么场景用,以及__init__.py暴露的那一排「运行器」(Agent / Team / Graph / Chat …)各自的定位。
本章是「地基」。管线内部每一步怎么跑,交给 第 2 章;模型、工具、安全各自成章(03/04/05)。读完本章,你应该能回答:「我 new 一个 Agent、new 一个 Task、调 do 会发生什么。」
1. 先建立心智模型:名词 vs 动词
Upsonic 的顶层设计可以用一句话概括:把「做什么」和「谁来做」彻底分开。
- Task = 名词(要做的事)。 它是一个纯数据对象——描述、工具清单、上下文、期望返回格式。它自己不会跑,不知道用哪个模型。
- Agent = 执行者(会做事的人)。 它挂着模型、记忆、安全策略、可靠性层。它知道怎么把一个 Task 跑完。
这种拆分带来一个直接好处:同一个 Task 可以丢给不同 Agent 跑,同一个 Agent 可以复用于很多 Task。 就像「工单」和「工人」分开——工单描述活儿,工人带着工具来干。
最小的一次使用长这样(源码 docstring 里的例子,agent/agent.py:167-175):
from upsonic import Agent, Task
agent = Agent("openai/gpt-4o") # 执行者:只给了模型
task = Task("What is 1 + 1?") # 要做的事:只给了一句描述
result = agent.do(task) # 把 task 喂给 agent 跑
Agent 和 Task 都从顶层包直接导出——但注意,upsonic/__init__.py 用的是懒加载:from upsonic import Agent 触发 __getattr__ → _get_Agent() 才真正 import 重模块(__init__.py:59-67, 105-136)。这是为了让 import upsonic 本身很快,不被一堆重依赖拖慢。
2. Task:一次「要做的事」
Task 是个 Pydantic 模型(tasks/tasks.py:21 class Task(BaseModel))。字段很多,但真正你会主动传的就那么几个。先看这几个核心的。
2.1 你最常传的四个字段
| 字段 | 类型 | 作用 | 源码 |
|---|---|---|---|
description | str(必填) | 这次要做什么, 自然语言 | tasks/tasks.py:26 |
tools | list[Any] | 这个 task 允许用的工具(函数 / ToolKit / MCP / 甚至别的 Agent) | tasks/tasks.py:28 |
context | Any | 参考材料:字符串、别的 Task、KnowledgeBase、文件/文件夹路径 | tasks/tasks.py:33 |
response_format | type | 期望返回的结构;默认 str,可传一个 Pydantic 模型强制结构化输出 | tasks/tasks.py:30 |
其余大量字段(缓存、策略作用域、向量检索参数、usage/run 状态…)大多有默认值,进阶才碰。下面挑最能体现设计意图的三处细讲。
2.2 response_format:想要结构化输出就传个模型
默认 response_format = str,即返回纯文本。想让模型吐结构化结果,传一个 Pydantic 类即可:
# 示意,非源码
from pydantic import BaseModel
class Invoice(BaseModel): # 定义你想要的形状
tax_number: str
total: float
task = Task(
"从这张发票里读出税号和总额",
context=["invoice.pdf"], # 文件路径直接塞进 context(见 2.3)
response_format=Invoice, # 要求返回 Invoice 而不是一段文本
)
response_format 允许 Pydantic 模型 / str / None 三种(tasks/tasks.py:30 的 Union[Type[BaseModel], type[str], None])。它怎么被翻译成模型端的「结构化输出」约束,是模型层的事,见 03-model-layer.md。
2.3 精华:context 里的文件路径会被自动抽成附件
这是 Task 一个很贴心的设计。你往 context 里塞的如果是一个存在的文件路径(甚至文件夹),Task 会在构造时把它从 context 里摘出来、转成 attachment,而不是把这串路径当普通文本喂给模型。
流程发生在 __init__ 里(tasks/tasks.py:365-374):
# tasks/tasks.py:365-374 (节选)
context, extracted_files = self._extract_files_from_context(context)
...
if extracted_files:
attachments.extend(extracted_files)
真正干活的是静态方法 _extract_files_from_context(tasks/tasks.py:211)。它的判定逻辑值得一看:
- 遍历
context(支持 list / dict / 单个字符串三种形态,tasks/tasks.py:235-301); - 每一项用
_is_file_path(tasks/tasks.py:104)判断是不是存在的文件——是,就摘去extracted_files; - 用
_is_folder_path(tasks/tasks.py:145)判断是不是文件夹——是,就用_get_files_from_folder(tasks/tasks.py:184)递归把夹子里所有文件都抽出来; - 其余的才留在「清洗后的 context」里,当作真正的文本上下文。
一句话看这张「谁去哪」的分流图:
context 里的每一项
│
▼
是存在的文件? ──是──▶ 移入 attachments(不再当文本)
│否
▼
是存在的文件夹? ─是──▶ 递归取夹内所有文件 ──▶ attachments
│否
▼
留在 cleaned_context(当文本上下文喂模型)
坑点(诚实提醒): 判定不只看「存不存在」。如果一个字符串长得像文件路径(以 .pdf/.md/.py 等结尾,或含 /、\)但文件其实不存在,_is_file_path 会直接抛 FileNotFoundError(tasks/tasks.py:132-136),再被 __init__ 包成一条更友好的错误(tasks/tasks.py:367-368)。也就是说:往 context 塞一个拼错的 .pdf 路径,Task(...) 构造当场就炸,而不是静默当文本。
抽出来的文件后续怎么用?attachments 会被 attachments_base64 属性(tasks/tasks.py:696)逐个读成 base64,交给多模态模型。
2.4 Task 也 承载「运行状态」
Task 不只是输入,它还是这次运行的状态载体——一次运行的结果和过程都记回同一个 task 对象上:
| 你能读到什么 | 属性 | 源码 |
|---|---|---|
| 最终输出 | task.response | tasks/tasks.py:741 |
| 这次跑的用量(token/成本/耗时) | task.usage | tasks/tasks.py:432 |
| 命中缓存了吗 | task.cache_hit | tasks/tasks.py:756 |
| 期间调了哪些工具 | task.tool_calls | tasks/tasks.py:766 |
| 运行状态(完成/暂停/取消/出错) | task.status | tasks/tasks.py:48 |
| 是否需要「续跑」而非重跑 | task.is_problematic | tasks/tasks.py:452 |
关键理解:「状态」和「配置」是分开管理的。 每次全新运行,管线第 0 步会调 task_start()(tasks/tasks.py:863),它先 reset_run_state()(tasks/tasks.py:820)把上一次的 response、tool_calls、缓存命中、策略标记等逐一清空,再新建 TaskUsage 开计时。但它绝不动你传的配置(description / tools / response_format)和持久字段(task_id_ / task_usage_id_)。这就是为什么同一个 task 对象可以安全地被重试、被续跑。
is_problematic(tasks/tasks.py:452)是这套状态机的关键判据:当 status 落在 paused / cancelled / error 三者之一,它返回 True,提示调用方——这次别用 do_async() 重头跑,要用 continue_run_async() 续跑(见 §4.3)。
3. Agent:一个「会做事的执行者」
Agent(agent/agent.py:150 class Agent(BaseAgent))是整个框架的集成中枢——模型、工具、记忆、上下文管理、缓存、安全策略、可靠性层、Canvas、可观测性,全挂在它身上。它的 __init__ 有 50+ 个参数(agent/agent.py:228-303),吓人,但绝大多数有默认值,Agent("openai/gpt-4o") 就能跑。
3.1 关键构造参数(按「关注点」分组)
下面把 __init__ 里最值得知道的参数按用途归类。完整清单看 agent/agent.py:228-303。
模型 / 身份
| 参数 | 默认 | 作用 |
|---|---|---|
model | "openai/gpt-4o" | provider/model 字符串或 Model 实例;__init__ 里立刻 infer_model(model) 解析成模型对象(agent/agent.py:385-387) |
name / role / goal / instructions | None | Agent 的身份与职责,会拼进 system prompt |
记忆 / 存储
| 参数 | 默认 | 作用 |
|---|---|---|
memory | None | 对话历史 / 用户画像的记忆实例 |
db | None | 数据库实例;若给了 db,它的 db.memory 会覆盖 memory(agent/agent.py:437-440) |
session_id / user_id | None | 会话与用户标识,记忆按此分桶 |
安全护栏(见 05-safety-engine.md)
| 参数 | 作用 |
|---|---|
user_policy / agent_policy | 对输入 / 对输出的安全策略;各自被包进一个 PolicyManager(agent/agent.py:508-522) |
tool_policy_pre / tool_policy_post | 工具执行前 / 后的策略,包进 ToolPolicyManager(agent/agent.py:534-537) |
*_policy_feedback | 命中策略时是「直接拦」还是「带反馈重试」 |
可靠性 / 反思 / 上下文治理
| 参数 | 默认 | 作用 |
|---|---|---|
reliability_layer | None | 可靠性层(校验 agent / 编辑 agent / 多轮质量改进) |
reflection / reflection_config | False | 自 我反思;reflection=True 但没给 config 时自动建默认 config(agent/agent.py:544-554) |
context_management | False | 开启后上下文接近模型上限时自动裁剪/摘要历史;构造时就装好中间件(agent/agent.py:451-463) |
工具 / 技能(见 04-tools-system.md)
| 参数 | 作用 |
|---|---|
tools | agent 级工具(函数 / ToolKit / 别的 agent);构造末尾 _register_agent_tools() 立即注册(agent/agent.py:576) |
skills | 技能包;其 get_tools() 会被并进 tools(agent/agent.py:484-485) |
enable_thinking_tool / enable_reasoning_tool | 开启编排式思考 / 推理工具 |
运行行为
| 参数 | 默认 | 作用 |
|---|---|---|
retry | 1 | 重试次数;< 1 直接抛 ValueError(agent/agent.py:467-468) |
mode | "raise" | 失败时抛异常还是返回 False;非法值抛错(agent/agent.py:469-470) |
tool_call_limit | 100 | 单次执行最多调多少次工具 |
print | None | 是否打印执行面板;和环境变量 UPSONIC_AGENT_PRINT 有优先级关系(agent/agent.py:431-433) |
3.2 BaseAgent:一个空基 类,专为打破循环依赖
Agent 继承自 BaseAgent,而 BaseAgent(agent/base.py:3)是个完全空的抽象类——pass 而已。
它存在的唯一理由写在 docstring 里:打破循环依赖。像 Graph 这种需要「持有某种 agent」的复杂组件,只要依赖这个轻量基类,就不必 import 沉重的 Agent 具体实现,从而避免 import 环。任何继承 BaseAgent 的类都能被塞进 Graph 当节点用。这是一个很典型的「用空基类当接口锚点」的工程手法。
3.3 构造时都发生了什么(不进管线)
Agent.__init__ 是纯装配,不跑任何任务。按顺序做的关键几件事:
infer_model(model) # 解析模型 agent.py:385-387
装可观测性 / promptlayer agent.py:389-391
db 优先于 memory agent.py:437-440
context_management 中间件(若开) agent.py:451-463
校验 retry / mode 合法 agent.py:467-470
skills.get_tools() 并入 tools agent.py:484-485
建 PolicyManager × 2 + ToolPolicyManager × 2 agent.py:508-537
reflection 默认 config / processor agent.py:544-554
建 CacheManager + ToolManager,注册 agent 级工具 agent.py:563-576
理解这一点很重要:Agent 一 new 出来就已经把所有子系统装好了,do 只是「按下开始」。agent_id / agent_usage_id 则是懒生成——第一次访问才 uuid4() / new_usage_id(agent/agent.py:776-792)。
4. 入口方法族:do 一家子
Agent 对外的「跑一个 task」入口有好几个,但它们全都收敛 到一个真正的核心:do_async(agent/agent.py:3649)。其余都是它的包装。
4.1 全景:谁调谁
do(task) ──────┐ 同步阻塞、不打印
│
print_do(task) ─┤──▶ _run_in_bg_loop(...) ──▶ do_async(...) ◀── 唯一核心
│ (丢进常驻后台事件循环) │
print_do_async ────┘ ▼
跑 24 步管线 PipelineManager
stream(task) ───▶ 起线程跑 astream， (见 02-execution-pipeline.md)
用队列把结果搬回同步侧
astream(task) ──▶ 独立的流式管线(不走 do_async)
关键点:同步方法不是自己实现的,而是把协程丢进一个进程级常驻的后台事件循环 _run_in_bg_loop(agent/agent.py:37-55)。为什么不用 asyncio.run()?docstring 说得很直白(agent/agent.py:14-18):asyncio.run() 每次会关掉事件循环,导致 OpenAI SDK 里缓存的 httpx.AsyncClient 连接失效;所以 Upsonic 维护一个常驻 loop,整个进程复用。
4.2 每个入口什么时候用
| 方法 | 同步/异步 | 打印面板 | 典型场景 | 源码 |
|---|---|---|---|---|
do | 同步 | 否(除非配置开) | 脚本里最常用,拿返回值 | agent/agent.py:4074 |
do_async | 异步 | 否 | 异步应用 / 并发跑多个 | agent/agent.py:3649 |
print_do | 同步 | 是 | 交互演示,想看指标面板 | agent/agent.py:4129 |
print_do_async | 异步 | 是 | 异步版演示 | agent/agent.py:4176 |
stream | 同步 | — | 边生成边打印文本(同步侧) | agent/agent.py:4267 |
astream | 异步 | — | 边生成边消费(异步流) | agent/agent.py:4341 |
continue_run | 同步 | — | HITL / 出错 / 取消后续跑 | agent/agent.py:4946 |
几个共同点,记住能省很多困惑:
- 入参可以是
str、Task、或它们的列表。 传str时内部_convert_to_task直接包成Task(description=...)(agent/agent.py:949-963);传多元素列表则逐个顺序执行、返回结果列表(do_async开头的_handle_task_list_async,agent/agent.py:3707)。 return_output=False(默认)返回内容本身(str / Pydantic 实例);return_output=True返回完整的AgentRunOutput(agent/agent.py:3677)。print的优先级:环境变量UPSONIC_AGENT_PRINT> Agent 构造参数print> 方法默认(do=不打印,print_do=打印),逻辑集中在agent/agent.py:431-433与_print_method_default的传递。
4.3 continue_run:HITL 与错误恢复的「续跑」
普通的 do 是「从头跑到尾」。但 Upsonic 支持跑到一半停下来等人——这就是 HITL(Human-in-the-Loop,人在环路)。continue_run(agent/agent.py:4946)就是从暂停点接着往下跑,而不是重来。
它覆盖的续跑场景(docstring agent/agent.py:4966-4972):
| 场景 | 怎么触发续跑 |
|---|---|
| 外部工具执行 | 把带上外部结果的 task 传回去 |
| 用户确认 | 批准 / 拒绝某个工具调用 |
| 用户输入 | 为工具调用补上字段值 |
| 持久化 / 错误恢复 | 传 run_id,从存储里加载状态续跑 |
| 取消后恢复 | 传 run_id 续跑 |
它和 do 的分工用 task.is_problematic(§2.4)判定:status 是 paused/cancelled/error → 用 continue_run;否则用 do。底层它靠 run_id 把一次运行的状态串起来——do_async 在跑之前会 register_run(run_id) 并把 run_id 写回 task(agent/agent.py:3784-3804),暂停时不清 run_id(agent/agent.py:3864-3865),这样才能被续上。
注意流式模式不支持 HITL。
astream的 docstring 明确写了(agent/agent.py:4355),遇到 problematic 任务时它只发一条 warning 就 return(agent/agent.py:4397-4402)——要 HITL 请走do_async/continue_run。
4.4 as_mcp:把这个 Agent 变成一台 MCP 服务器
as_mcp(agent/agent.py:4214)是个有意思的「反向」入口:它把当前 Agent 包装成一个 MCP(Model Context Protocol)服务器,对外暴露一个 do 工具,别的系统就能像调用任何 MCP 工具那样把任务丢给这个 agent。
# 示意,非源码
server = agent.as_mcp(name="我的翻译助手")
server.run() # 起 stdio / sse / http 服务,别人就能连过来用
实现上它用 fastmcp 建一个 FastMCP 实例,注册一个内部 do(task: str) 工具,函数体里转调 agent.print_do(task)(agent/agent.py:4257-4263)。fastmcp 没装时会给出友好的安装提示(agent/agent.py:4230-4238)。这条线让 Upsonic 的 agent 既能用 MCP 工具(见 04),也能被当成 MCP 工具——双向打通。
5. 一排「运行器」:__init__.py 暴露了什么
顶层 __init__.py 的 __all__(__init__.py:145-159)导出了不止 Agent。它其实提供了一族不同抽象层级的运行器,对应不同复杂度的编排需求。选哪个,取决于你要跑的是「一个 agent」「一队 agent」还是「一张流程图」。
| 运行器 | 一句话定位 | 何时选它 | 源码锚点 |
|---|---|---|---|
Agent(别名 Clanker) | 全功能单 agent,一切的中心 | 绝大多数场景 | agent/agent.py:150;Clanker = Agent 在 agent/agent.py:5424 |
AutonomousAgent | Agent 子类,预装文件系统 + shell + 沙箱工作区 | 编码助手 / DevOps / 需要读写文件和跑命令 | agent/autonomous_agent/autonomous_agent.py:30 |
Team | 多 agent 协作,支持嵌套团队 | 分层多智能体、任务分派 | team/team.py:20 |
Graph | 用「节点 + 状态」把多个 task/agent 编成工作流,支持并行 | 有明确 DAG / 条件流转的流程 | graph/graph.py:392 |
Chat | 有状态的会话编排器,绑存储、管记忆与成本 | 多轮对话式应用 | chat/chat.py:40 |
Direct | 极简高速接口,无记忆 / 无 KB / 无工具,只求快 | 纯抽取 / 一次性结构化调用 | direct.py:14 |
RalphLoop | 自主开发循环:无限轮里每轮起全新 Agent 跑一个 task | 长时自主编码(见 06) | ralph/loop.py:24 |
Simulation | 场景模拟 | 评测 / 演练 | simulation/simulation.py |
怎么理解这张表?按「抽象层级从低到高」排:
Direct 只想要一次快速 LLM 调用,不要花活
│
Agent 一个完整 agent(工具/记忆/安全/可靠性) ← 本章主角
│
AutonomousAgent Agent + 现成的文件/shell 能力
│
Chat 把 Agent 包成有状态的多 轮会话
│
Team / Graph 多个 agent / task 的编排(队列式 / 图式)
│
RalphLoop 用「每轮全新 Agent」堆出长时自主开发
Direct vs Agent 的取舍很能说明设计意图:Direct 的 docstring 直说它「focuses on maximum speed and direct data retrieval」,刻意砍掉记忆、KB、工具(direct.py:14-19)——当你只想「读个 PDF 返回一个字段」时,Agent 那套管线是浪费,Direct 才对。
AutonomousAgent 则是「加料的 Agent」:它继承 Agent(autonomous_agent.py:30),默认塞好 InMemoryStorage + Memory + 文件工具 + shell 工具 + 工作区沙箱,并把 print 默认改成 True(autonomous_agent.py:34-46)。想要一个「开箱能读写代码」的 agent,直接用它,别自己攒。
6. 边界与局限(诚实说明)
Task的文件判定会「猜」路径:字符串以已知扩展名结尾、或含/`` 却不存在时会抛FileNotFoundError(tasks/tasks.py:132-136)。好处是早失败,代价是——一段碰巧长得像路径的普通文本可能被误判。这是刻意的严格取舍。- 流式不支持 HITL:
astream遇到需要人类介入的任务只会 warning 后 return(agent/agent.py:4397-4402)。 - 同步入口依赖常驻后台 loop:
do/print_do/stream都把活儿丢给进程级的_bg_loop(agent/agent.py:17-34)。这是为连接复用做的取舍,但也意味着有一个常驻守护线程。 - 本章不覆盖管线内部:
do_async之后那 24 步(InitializationStep…CallManagementStep,agent/agent.py:4650-4675)怎么跑,是 第 2 章 的事。
7. 代码地图(导航索引)
按「符号名」定位比行号更抗漂移(上游更新后行号会变,符号名通常还在)。
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| Task 数据模型 | src/upsonic/tasks/tasks.py | Task |
| context 里抽文件 | src/upsonic/tasks/tasks.py | Task._extract_files_from_context / _is_file_path / _is_folder_path / _get_files_from_folder |
| 每次运行清状态 | src/upsonic/tasks/tasks.py | Task.reset_run_state / Task.task_start |
| 需不需要续跑的判据 | src/upsonic/tasks/tasks.py | Task.is_problematic |
| Agent 集成中枢 | src/upsonic/agent/agent.py | Agent.__init__ |
| 空基类(破循环依赖) | src/upsonic/agent/base.py | BaseAgent |
| 核心异步入口 | src/upsonic/agent/agent.py | Agent.do_async |
| 同步入口 | src/upsonic/agent/agent.py | Agent.do / Agent.print_do |
| 常驻后台事件循环 | src/upsonic/agent/agent.py | _get_bg_loop / _run_in_bg_loop |
| 流式入口 | src/upsonic/agent/agent.py | Agent.stream / Agent.astream |
| HITL 续跑 | src/upsonic/agent/agent.py | Agent.continue_run / Agent.continue_run_async |
| 暴露为 MCP 服务器 | src/upsonic/agent/agent.py | Agent.as_mcp |
| 24 步管线清单 | src/upsonic/agent/agent.py | Agent._create_direct_pipeline_steps |
| 顶层导出与懒加载 | src/upsonic/__init__.py | __getattr__ / _get_Agent / __all__ |
| 加料 agent | src/upsonic/agent/autonomous_agent/autonomous_agent.py | AutonomousAgent |
| 极简高速接口 | src/upsonic/direct.py | Direct |
| 多 agent / 工作流 / 会话 | src/upsonic/team/team.py、graph/graph.py、chat/chat.py | Team / Graph / Chat |
| 自主开发循环 | src/upsonic/ralph/loop.py | RalphLoop |
下一步: 想知道 agent.do(task) 之后那条管线逐步发生了什么(记忆加载 → 建 prompt → 装工具 → 调模型 → 处理响应 → 存记忆…),读 02-execution-pipeline.md。