Agent¶
Agent 是 Agentica 的核心组件——它将模型、工具和记忆连接在一起,能够思考、决策和执行动作。
基本结构¶
Agent
├── Model # 大脑:LLM 提供推理能力
├── Tools # 双手:与外部世界交互的能力
├── Memory # 运行时记忆:当前会话消息历史
├── Workspace # 持久化记忆:跨会话长期记忆
├── Knowledge # 知识库:外部文档检索 (RAG)
├── Instructions # 指令:定义 Agent 的行为和风格
└── Runner # 执行引擎(自动创建,无需手动管理)
创建 Agent¶
最小示例¶
from agentica import Agent, ZhipuAI
agent = Agent(model=ZhipuAI())
result = agent.run_sync("一句话介绍北京")
print(result.content)
完整配置示例¶
import asyncio
from agentica import Agent, OpenAIChat
from agentica.tools.duckduckgo_tool import DuckDuckGoTool
from agentica.tools.url_crawler_tool import UrlCrawlerTool
from agentica.agent.config import PromptConfig, ToolConfig, WorkspaceMemoryConfig
from agentica import Workspace
async def main():
agent = Agent(
# 核心定义
name="Research Assistant",
model=OpenAIChat(id="gpt-4o"),
instructions=[
"你是一个专业的研究助手,擅长收集和整理最新信息",
"搜索时使用多个关键词进行验证,确保信息准确",
"用中文回答,条理清晰,附上信息来源",
],
tools=[DuckDuckGoTool(), UrlCrawlerTool()],
# Prompt 配置
prompt_config=PromptConfig(
markdown=True, # 输出 Markdown 格式
add_datetime_to_instructions=True, # 注入当前日期
),
# 工具配置
tool_config=ToolConfig(
tool_call_limit=20, # 单次最多调用 20 次工具
compress_tool_results=True, # 自动压缩大工具结果
),
# 长期记忆
workspace=Workspace("./workspace"),
long_term_memory_config=WorkspaceMemoryConfig(
load_workspace_memory=True,
max_memory_entries=5,
),
# 会话历史
add_history_to_messages=True,
history_window=5,
)
result = await agent.run("2025 年 AI 领域最重要的进展有哪些?")
print(result.content)
asyncio.run(main())
参数分层¶
第一层:核心定义¶
| 参数 | 类型 | 说明 |
|---|---|---|
model |
Model |
LLM 模型实例(必填) |
name |
str |
Agent 名称,注入到 System Prompt |
description |
str |
Agent 描述,作为 System Prompt 开头 |
instructions |
str \| List[str] \| Callable |
行为指令,支持动态计算 |
tools |
List[Tool \| Callable] |
工具列表 |
knowledge |
Knowledge |
RAG 知识库 |
team |
List[Agent] |
团队成员(委派任务给其他 Agent) |
workspace |
Workspace |
持久化工作空间 |
session_id |
str |
会话 ID(用于会话持久化和恢复) |
response_model |
Type[BaseModel] |
结构化输出(Pydantic 模型) |
第二层:会话历史¶
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
add_history_to_messages |
bool |
False |
将历史消息加入上下文 |
history_window |
int |
3 |
保留最近 N 轮历史 |
structured_outputs |
bool |
False |
严格 JSON 结构化输出(部分 API 支持) |
debug |
bool |
False |
启用调试日志 |
第三层:打包配置¶
| 参数 | 类型 | 说明 |
|---|---|---|
prompt_config |
PromptConfig |
提示词工程配置 |
tool_config |
ToolConfig |
工具调用行为配置 |
long_term_memory_config |
WorkspaceMemoryConfig |
工作空间记忆配置 |
team_config |
TeamConfig |
多 Agent 团队协作配置 |
运行方式¶
Agentica 采用 async-first 架构:
import asyncio
from agentica import Agent, ZhipuAI
agent = Agent(model=ZhipuAI())
# 1. 异步非流式(推荐,生产环境使用)
async def example_async():
result = await agent.run("你好")
print(result.content)
# 2. 异步流式(实时显示,长响应适用)
async def example_stream():
async for chunk in agent.run_stream("写一首诗"):
if chunk.content:
print(chunk.content, end="", flush=True)
# 3. 同步非流式(脚本/Jupyter 适用)
result = agent.run_sync("你好")
print(result.content)
# 4. 同步流式(Jupyter / 非 async 环境)
for chunk in agent.run_stream_sync("写一首诗"):
if chunk.content:
print(chunk.content, end="", flush=True)
asyncio.run(example_async())
| 方法 | 类型 | 适用场景 |
|---|---|---|
run(msg) |
async |
FastAPI、异步服务、生产环境 |
run_stream(msg) |
async generator |
实时流式响应 |
run_sync(msg) |
sync |
脚本、Jupyter、CLI |
run_stream_sync(msg) |
sync iterator |
同步代码中的流式输出 |
print_response(msg) |
async |
快速测试(自带格式化打印) |
Instructions(指令)¶
指令定义 Agent 的行为、风格和约束。
静态指令¶
agent = Agent(
instructions=[
"你是一个 Python 代码审查专家",
"重点检查:安全漏洞、性能瓶颈、代码可读性",
"发现问题时,提供具体的修复建议和代码示例",
"使用中文回复,格式清晰",
],
)
动态指令(基于上下文计算)¶
from datetime import date
def get_instructions(agent) -> list:
"""运行时动态计算指令"""
instructions = ["你是一个智能助手"]
today = date.today()
if today.weekday() >= 5: # 周末
instructions.append("今天是周末,优先考虑娱乐和生活类问题")
if agent.session_state.get("expert_mode"):
instructions.append("使用专业术语,提供深度分析")
return instructions
agent = Agent(instructions=get_instructions)
# 每次 run() 前,Callable 会被调用并传入 agent 实例
完整 System Prompt¶
如果需要完全控制 System Prompt(绕过 Agentica 的 prompt 组装逻辑):
from agentica.agent.config import PromptConfig
agent = Agent(
prompt_config=PromptConfig(
system_prompt="你是一个友好的中文助手。回答要简洁准确,控制在 200 字以内。",
),
)
# 也可以是 Callable,运行时动态计算:
agent = Agent(
prompt_config=PromptConfig(
system_prompt=lambda agent: f"你是 {agent.name},今天是{date.today()}。",
),
)
PromptConfig(进阶)¶
PromptConfig 控制 System Prompt 的构建细节:
from agentica.agent.config import PromptConfig
agent = Agent(
model=ZhipuAI(),
prompt_config=PromptConfig(
# 内容补充(追加到 instructions 后)
task="分析用户提供的数据并给出洞察",
role="数据分析师",
guidelines=["使用统计方法验证结论", "图表描述要详细"],
expected_output="分析报告:问题→数据→结论→建议",
# 格式控制
markdown=True, # 要求 LLM 输出 Markdown
output_language="中文", # 强制中文输出
add_datetime_to_instructions=True, # 注入当前日期
# 安全开关
prevent_hallucinations=True, # 注入"不确定时说不知道"指令
prevent_prompt_leakage=True, # 注入"不泄露 system prompt"指令
limit_tool_access=True, # 注入"只使用已提供的工具"指令
# Agentic Prompt 模式(见下文)
enable_agentic_prompt=False, # 默认关闭
# 极简模式(最低 token 消耗)
minimal=False,
),
)
enable_agentic_prompt:自主 Agent 模式¶
开启后,System Prompt 注入四个行为模块,让 Agent 具备自主完成复杂任务的能力:
from agentica.agent.config import PromptConfig
agent = Agent(
model=OpenAIChat(id="gpt-4o"),
prompt_config=PromptConfig(enable_agentic_prompt=True),
)
注入的四个模块:
| 模块 | 文件 | 效果 |
|---|---|---|
| Soul | soul.md |
核心行为准则:技术准确性优先、有主见、避免过度工程 |
| Tools Guide | tools.md |
工具使用策略:禁用 cat/grep shell 等效命令,强制使用内置工具 |
| Heartbeat | heartbeat.md |
迭代控制:必须持续直到问题彻底解决,不半途而废 |
| Self Verification | self_verification.md |
代码变更后必须运行 lint/test 验证 |
适用场景:DeepAgent(CLI 模式)默认开启;普通 Agent 默认关闭。开启后 Agent 更擅长多步骤代码任务,但 prompt token 会增加约 800 tokens。
minimal 模式¶
最低 token 消耗,跳过所有 prompt 组装,只输出一行系统提示:
agent = Agent(
model=ZhipuAI(),
prompt_config=PromptConfig(minimal=True),
)
# System Prompt: "You are Agent-Name, an AI assistant.\nCWD: ...\nDate: ..."
# 约 20 tokens,适合大批量、成本敏感的场景
也可通过环境变量启用:AGENTICA_SIMPLE=1 python script.py
ToolConfig(工具配置)¶
from agentica.agent.config import ToolConfig
agent = Agent(
model=ZhipuAI(),
tools=[...],
tool_config=ToolConfig(
tool_call_limit=30, # 单次 run 最多调用工具 30 次(防止无限循环)
compress_tool_results=True, # 大工具结果自动压缩(节省 token)
context_overflow_threshold=0.8, # 上下文达 80% 时触发压缩
max_repeated_tool_calls=3, # 相同工具+参数出现 3 次→注入"你在循环"提示
search_knowledge=True, # 允许 Agent 主动搜索知识库
),
)
WorkspaceMemoryConfig(长期记忆配置)¶
from agentica.agent.config import WorkspaceMemoryConfig
from agentica import Workspace
agent = Agent(
workspace=Workspace("./workspace"),
long_term_memory_config=WorkspaceMemoryConfig(
load_workspace_context=True, # 加载 AGENTS.md/PERSONA.md 等上下文文件
load_workspace_memory=True, # 加载相关记忆到 System Prompt
max_memory_entries=5, # 最多注入 5 条最相关的记忆
auto_archive=True, # 每次 run 后自动归档对话(无 LLM 成本)
auto_extract_memory=False, # 自动提取关键信息存记忆(有 LLM 成本)
),
)
RunResponse¶
run() 返回 RunResponse 对象,包含完整的执行信息:
from agentica import Agent, ZhipuAI, RunEvent
agent = Agent(model=ZhipuAI())
result = await agent.run("分析这段代码的时间复杂度")
# 基础内容
result.content # str 或 BaseModel(结构化输出时)
result.model # 使用的模型 ID
result.run_id # 本次运行的唯一 ID
result.agent_id # Agent 的唯一 ID
# 性能指标
result.metrics # dict,包含 token 用量、延迟等
result.metrics["time"] # 总耗时(秒)
result.metrics["prompt_tokens"] # 输入 token 数
result.metrics["completion_tokens"] # 输出 token 数
# 工具调用信息
result.tool_calls # List[ToolCallInfo]
for tc in result.tool_calls:
print(f"{tc['tool_name']}: {tc['tool_call_error']}")
# 推理过程(DeepSeek-R1 等推理模型)
result.reasoning_content # str,LLM 的思考链
流式事件¶
from agentica import RunEvent
async for chunk in agent.run_stream("帮我写一个排序算法"):
match chunk.event:
case RunEvent.run_response:
print(chunk.content, end="", flush=True) # 正常输出 token
case RunEvent.tool_call_started:
print(f"\n🔧 调用工具: {chunk.content}")
case RunEvent.tool_call_completed:
print(f" ✓")
case RunEvent.run_completed:
print(f"\n完成 | tokens: {chunk.metrics}")
DeepAgent¶
DeepAgent 是预配置的高级 Agent,专为复杂的代码和研究任务设计,内置 20+ 工具并开启 enable_agentic_prompt=True:
from agentica import DeepAgent, OpenAIChat
agent = DeepAgent(
model=OpenAIChat(id="gpt-4o"),
name="My AI Assistant",
work_dir="./my-project", # 文件操作基准目录
workspace_path="./workspace", # 持久化记忆目录
)
result = await agent.run("""
分析当前项目的代码结构,找出潜在的性能瓶颈,
并为最重要的问题提供优化建议和修改后的代码
""")
DeepAgent 内置工具(ls, read_file, write_file, edit_file, multi_edit_file, glob, grep, execute, web_search, fetch_url, write_todos, task, user_input, save_memory, search_memory)
DeepAgent 默认开启:
- enable_agentic_prompt=True(Soul + Tools Guide + Heartbeat + Self Verification)
- compress_tool_results=True(大工具结果自动压缩)
- context_overflow_threshold=0.8(上下文 80% 触发压缩)
- max_repeated_tool_calls=3(循环检测)
- Workspace 长期记忆
- Session Log 会话持久化
结构化输出¶
通过 Pydantic 模型获取严格结构化的 JSON 数据:
import asyncio
from pydantic import BaseModel, Field
from typing import List
from agentica import Agent, ZhipuAI
class CodeReview(BaseModel):
summary: str = Field(description="代码总体评价")
issues: List[str] = Field(description="发现的问题列表")
suggestions: List[str] = Field(description="改进建议")
score: int = Field(description="代码质量评分 1-10", ge=1, le=10)
async def main():
agent = Agent(
model=ZhipuAI(),
response_model=CodeReview,
)
result = await agent.run("def add(a, b): return a+b")
review: CodeReview = result.content # 自动解析为 CodeReview 对象
print(f"评分: {review.score}/10")
for issue in review.issues:
print(f" - {issue}")
asyncio.run(main())
多轮对话¶
import asyncio
from agentica import Agent, ZhipuAI
async def main():
agent = Agent(
model=ZhipuAI(),
add_history_to_messages=True, # 开启会话历史
history_window=10, # 保留最近 10 轮
)
# 第一轮
r1 = await agent.run("我叫小明,是一名 Python 开发者")
print(r1.content)
# 第二轮(Agent 记得上一轮的内容)
r2 = await agent.run("你还记得我的名字吗?")
print(r2.content) # "你叫小明,是一名 Python 开发者"
asyncio.run(main())
会话持久化(Session Log)¶
通过 session_id 参数启用 JSONL 会话日志,支持跨进程恢复:
import asyncio
from agentica import Agent, ZhipuAI
SESSION_ID = "my-research-session-001"
async def first_run():
agent = Agent(model=ZhipuAI(), session_id=SESSION_ID)
# 消息自动写入 ~/.agentica/projects/<cwd>/<session_id>.jsonl
result = await agent.run("我在研究机器学习优化算法")
print(result.content)
async def second_run():
# 用相同 session_id 创建,自动恢复上次会话
agent = Agent(model=ZhipuAI(), session_id=SESSION_ID)
result = await agent.run("继续上次的话题")
print(result.content) # Agent 记得上次的内容
asyncio.run(first_run())
asyncio.run(second_run())
clone():并行 Agent¶
clone() 创建当前 Agent 的副本,重置运行时状态(run_id、消息历史、_running 标记),同时给每个 clone 独立的 Model 实例,适合并发场景:
import asyncio
from agentica import Agent, OpenAIChat
async def main():
base = Agent(
model=OpenAIChat(id="gpt-4o"),
instructions=["你是一个代码审查专家"],
)
# 并发处理多个文件
files = ["auth.py", "database.py", "api.py"]
tasks = [
base.clone().run(f"审查 {f}")
for f in files
]
results = await asyncio.gather(*tasks)
for f, r in zip(files, results):
print(f"=== {f} ===\n{r.content}")
asyncio.run(main())
System Prompt 三区结构¶
Agentica 将 System Prompt 分为三个区域,最大化 LLM prefix-cache 命中率,节省 API 费用:
┌─ STATIC ZONE(不变,始终可被 cache)─────────────────────┐
│ description, instructions, guidelines, expected_output │
│ → 固定内容,跨 run 完全相同 │
├─ SEMI-STATIC ZONE(少变)───────────────────────────────┤
│ workspace context (AGENTS.md 等), git status │
│ → 文件内容变化时才更新 │
├─ DYNAMIC ZONE(每轮可变,必须在最后)────────────────────┤
│ workspace memory(相关记忆),session summary │
│ datetime(日期,精确到天) │
│ → 每次 run 可能不同 │
└─────────────────────────────────────────────────────────┘
日期精确到天(YYYY-MM-DD),同一天的请求 datetime 字段完全相同,前缀 cache 可复用。
SandboxConfig(沙箱安全)¶
对不受信任的代码执行场景,启用沙箱限制:
from agentica.agent.config import SandboxConfig
from agentica import DeepAgent, ZhipuAI
agent = DeepAgent(
model=ZhipuAI(),
sandbox_config=SandboxConfig(
enabled=True,
writable_dirs=["./output"], # 只允许写这个目录
allowed_commands=["python", "pip"], # 只允许运行 Python 和 pip
max_execution_time=60, # 命令最长执行 60 秒
),
)
沙箱说明
SandboxConfig 是尽力而为的安全屏障,不是真正的系统级沙箱。对不受信任的代码,应使用 Docker/seccomp 等 OS 级隔离。
下一步¶
- Team -- 多 Agent 委派与协作
- Workflow -- 确定性工作流编排
- 工具系统 -- 工具开发和配置
- Memory & Workspace -- 长期记忆机制
- Hooks -- 生命周期钩子
- Guardrails -- 输入/输出安全守卫
- RunConfig -- 运行时配置(Cost Budget、超时)
- API 参考 -- 完整 API 文档