AgentOS 依赖传递是一种强大的机制,允许在运行时向 Agent 注入动态上下文和参数。通过这种方式,您可以创建更加灵活、可配置的 Agent 系统,适应各种复杂的业务场景。
依赖是指在 Agent 执行过程中需要的动态参数或上下文信息。依赖可以是:
模板变量是指在 Agent 指令或用户消息中使用的占位符,格式为 {variable_name}。这些变量会在运行时被依赖中的对应值替换。
运行上下文是存储依赖和其他运行时信息的对象,贯穿整个 Agent 运行过程。
让我们详细分析 pass_dependencies_to_agent.py 示例代码:
"""Example for AgentOS to show how to pass dependencies to an agent."""
from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.os import AgentOS
# Setup the database
db = PostgresDb(id="basic-db", db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")
# Setup basic agents, teams and workflows
story_writer = Agent(
id="story-writer-agent",
name="Story Writer Agent",
db=db,
markdown=True,
instructions="You are a story writer. You are asked to write a story about a robot. Always name the robot {robot_name}",
)
# Setup our AgentOS app
agent_os = AgentOS(
description="Example AgentOS to show how to pass dependencies to an agent",
agents=[story_writer],
)
app = agent_os.get_app()
if __name__ == "__main__":
"""Run your AgentOS.
Test passing dependencies to an agent:
curl --location 'http://localhost:7777/agents/story-writer-agent/runs' \
--header 'Content-Type: application/x-www-form-urlencoded' \
--data-urlencode 'message=Write me a 5 line story.' \
--data-urlencode 'dependencies={"robot_name": "Anna"}'
"""
agent_os.serve(app="pass_dependencies_to_agent:app", reload=True)
db = PostgresDb(id="basic-db", db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")
id 参数用于标识数据库实例db_url 参数指定了数据库连接 URLstory_writer = Agent(
id="story-writer-agent",
name="Story Writer Agent",
db=db,
markdown=True,
instructions="You are a story writer. You are asked to write a story about a robot. Always name the robot {robot_name}",
)
id 和 name 参数用于标识和命名 Agentdb 参数指定了 Agent 使用的数据库markdown=True 表示 Agent 的输出将使用 Markdown 格式instructions 参数包含了 Agent 的核心指令,其中 {robot_name} 是一个模板变量,将在运行时被替换agent_os = AgentOS(
description="Example AgentOS to show how to pass dependencies to an agent",
agents=[story_writer],
)
app = agent_os.get_app()
description 参数描述了应用的功能agents 参数指定了应用中包含的 Agent 列表get_app() 方法返回一个 FastAPI 应用实例,用于启动 Web 服务if __name__ == "__main__":
agent_os.serve(app="pass_dependencies_to_agent:app", reload=True)
app 参数指定了要启动的应用reload=True 表示在开发模式下自动重载代码curl --location 'http://localhost:7777/agents/story-writer-agent/runs' \
--header 'Content-Type: application/x-www-form-urlencoded' \
--data-urlencode 'message=Write me a 5 line story.' \
--data-urlencode 'dependencies={"robot_name": "Anna"}'
这个 curl 命令演示了如何通过 API 向 Agent 传递依赖:
message 参数包含了用户请求dependencies 参数包含了要传递的依赖,这里是 {"robot_name": "Anna"}RunContext 对象dependencies 中的值存储到 RunContext 中RunContext 中RunContext 中的依赖值替换 Agent 指令和用户消息中的模板变量"You are a story writer. You are asked to write a story about a robot. Always name the robot {robot_name}"
- 替换后:"You are a story writer. You are asked to write a story about a robot. Always name the robot Anna"
最常见的依赖类型是基本数据类型,如字符串、数字等:
# API 调用示例
dependencies={"robot_name": "Anna", "story_length": 5}
# 对应的 Agent 指令
instructions="You are a story writer. Write a {story_length} line story about a robot named {robot_name}"
依赖也可以是复杂数据类型,如字典或列表:
# API 调用示例
dependencies={
"robot": {
"name": "Anna",
"type": "service robot",
"color": "blue"
},
"story_elements": ["adventure", "friendship", "discovery"]
}
# 对应的 Agent 指令
instructions="You are a story writer. Write a story about a {robot.type} named {robot.name} with the following elements: {story_elements}"
依赖可以是函数或方法,这些函数会在运行时被执行,其结果会被用作依赖值:
# 定义一个依赖函数
def get_current_time(run_context):
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 代码调用示例
agent.run(
"Write a story about a robot created on {creation_date}",
dependencies={"creation_date": get_current_time}
)
当依赖是可调用对象时,Agent 会根据其签名自动注入以下参数:
agent:当前 Agent 实例run_context:当前运行上下文对象全局依赖是在 Agent 初始化时定义的,对所有运行有效:
story_writer = Agent(
id="story-writer-agent",
instructions="You are a story writer. Always name the robot {robot_name}",
dependencies={"robot_name": "DefaultRobot"} # 全局依赖
)
局部依赖是在运行时传递的,仅对当前运行有效,会覆盖全局依赖:
# API 调用示例,局部依赖会覆盖全局依赖
dependencies={"robot_name": "Anna"}
Agent 提供了两种依赖解析方法:
_resolve_run_dependencies - 用于同步执行场景_aresolve_run_dependencies - 用于异步执行场景RunContext 中RunContext 中模板替换是通过 Python 的字符串格式化功能实现的,使用 str.format(**dependencies) 方法将模板变量替换为实际值。
当依赖是可调用对象时,Agent 会使用反射机制分析其签名,并自动注入以下参数:
agent:当前 Agent 实例run_context:当前运行上下文对象选择清晰、描述性的键名,便于理解和维护:
# 好的做法
dependencies={"robot_name": "Anna", "story_genre": "science fiction"}
# 不好的做法
dependencies={"rn": "Anna", "sg": "scifi"}
依赖的解析顺序是不确定的,因此不建议在依赖函数中引用其他依赖项:
# 不推荐的做法
dependencies={
"robot_name": "Anna",
"story_title": lambda run_context: f"The Adventures of {run_context.dependencies['robot_name']}" # 依赖其他依赖项
}
# 全局依赖用于通用配置
story_writer = Agent(
instructions="You are a {role} writing {genre} stories",
dependencies={"role": "story writer", "genre": "science fiction"} # 全局依赖
)
# 局部依赖用于动态值
agent.run(
"Write a story about a robot",
dependencies={"robot_name": "Anna"} # 局部依赖
)
依赖函数可能会抛出异常,确保有适当的错误处理:
def get_robot_name(run_context):
try:
# 可能会失败的操作
return get_robot_name_from_external_service()
except Exception as e:
# 提供默认值或处理错误
return "DefaultRobot"
为依赖函数添加类型注解,提高代码可读性和可维护性:
def get_current_time(run_context: RunContext) -> str:
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
问题:模板变量没有被正确替换
解决方案:
问题:依赖函数抛出异常
解决方案:
问题:依赖之间存在相互依赖,导致解析失败
解决方案:
在多 Agent 协作场景中,可以使用依赖传递机制在 Agent 之间共享上下文:
# 创建多个 Agent
story_writer = Agent(
id="story-writer-agent",
instructions="You are a story writer. Write a story about {topic}",
)
story_editor = Agent(
id="story-editor-agent",
instructions="You are a story editor. Edit the following story: {story}",
)
# 在工作流中传递依赖
workflow = Workflow(
id="story-creation-workflow",
steps=[
Step(agent=story_writer, output_key="story"),
Step(agent=story_editor, dependencies={"story": "{{steps.story.output}}"})
]
)
可以使用依赖传递机制动态配置 Agent 的行为:
# 创建一个可配置的 Agent
configurable_agent = Agent(
id="configurable-agent",
instructions="You are a {role}. {additional_instructions}",
)
# 根据不同场景传递不同的配置
agent.run(
"Perform the requested task",
dependencies={
"role": "customer service representative",
"additional_instructions": "Be polite and helpful"
}
)
agent.run(
"Perform the requested task",
dependencies={
"role": "technical support engineer",
"additional_instructions": "Provide detailed technical information"
}
)
AgentOS 依赖传递机制是一个强大的工具,允许您创建灵活、可配置的 Agent 系统。通过合理使用依赖传递,您可以:
还没有人回复