Agno 依赖注入与可调用工厂模式

Agno 依赖注入与可调用工厂模式

学习目标

本章将分析 Agno 的运行时依赖解析和工厂模式：

• dependencies dict 的自动签名检测与参数注入
• Callable Factory 模式实现运行时动态工具列表
• 缓存机制防止重复实例化
• 知识检索的可替换函数

项目实践

依赖注入系统

Agno 使用 dependencies dict 注入运行时数据到工具和钩子中：

# 伪代码：依赖定义
agent = Agent(
    dependencies={
        "db_client": PostgresClient(db_url="..."),
        "user_preferences": get_user_prefs(user_id),
        "config": app_config,
    },
)

依赖在管线步骤 3 解析，注入到需要它们的函数中：

# 伪代码：resolve_run_dependencies
def resolve_run_dependencies(agent: Agent, run_context: RunContext) -> None:
    for key, value in run_context.dependencies.items():
        if callable(value):
            sig = signature(value)
            kwargs = {}
            if "agent" in sig.parameters:
                kwargs["agent"] = agent
            if "run_context" in sig.parameters:
                kwargs["run_context"] = run_context
            result = value(**kwargs)
            run_context.dependencies[key] = result
        else:
            run_context.dependencies[key] = value

关键设计：

• 使用 inspect.signature 检测函数签名
• 自动匹配 agent 和 run_context 参数名并注入
• callable 的返回值替换原始值（缓存结果）

Callable Factory 模式

tools 字段支持运行时动态生成：

# 伪代码：动态工具工厂
def get_context_aware_tools(agent: Agent, run_context: RunContext) -> List:
    """根据上下文动态生成工具列表"""
    tools = []

    # 根据用户权限决定工具
    if run_context.session_state.get("admin"):
        tools.append(AdminTool())

    # 根据数据库连接生成查询工具
    db_name = run_context.session_state.get("db_name")
    if db_name:
        tools.append(QueryTool(db=db_name))

    return tools

agent = Agent(tools=get_context_aware_tools, cache_callables=True)

缓存机制

agent = Agent(
    tools=get_dynamic_tools,
    cache_callables=True,  # 缓存工厂结果
)

cache_callables=True 缓存 factory 的返回值，避免同一 Agent 的多次 run 重复调用 factory。

自定义缓存键：

def cache_key(agent: Agent, run_context: RunContext) -> Optional[str]:
    return f"tools_{run_context.session_state.get('db_name', 'default')}"

agent = Agent(
    tools=get_dynamic_tools,
    cache_callables=True,
    callable_tools_cache_key=cache_key,  # 自定义缓存键
)

知识检索的可替换函数

类似地，知识检索也可以替换：

# 伪代码：自定义知识检索
def custom_retriever(agent: Agent, query: str, num_documents: Optional[int], **kwargs):
    # 可以是外部 API 调用、缓存查询、启发式规则等
    return [{"content": "...", "source": "..."}]

agent = Agent(
    knowledge=my_knowledge,
    knowledge_retriever=custom_retriever,
    callable_knowledge_cache_key=lambda a, q, n: f"knowledge_{q}",
)

依赖在工具中的使用

工具可以通过 dependencies 获取运行时数据：

# 伪代码：工具使用依赖
@tool
def search_database(query: str, dependencies: Dict[str, Any]) -> str:
    db_client = dependencies.get("db_client")
    return db_client.search(query)

问题与规避

1. 依赖名称约定

依赖注入基于参数名匹配（agent、run_context）。如果工具函数的参数名不匹配，依赖不会被注入。

规避：文档化约定参数名。使用类型注解辅助 IDE 提示。

2. 缓存过期

cache_callables=True 缓存 factory 结果。如果外部状态变化，缓存可能过期。

规避：

• 使用自定义 cache_key 函数，在状态变化时返回不同的键
• 在关键位置检查缓存有效性
• 必要时设置 cache_callables=False

3. Callable Factory 的异常处理

Factory 函数可能抛出异常（如数据库连接失败）。

规避：

• Factory 函数做好异常处理，返回空列表而非抛出
• 在 resolve_run_dependencies 中捕获异常并记录日志

设计取舍

为什么用参数名匹配而非类型注解？

优势：简单直接，不需要类型匹配基础设施 代价：参数名冲突可能导致注入错误 替代方案：基于类型注解匹配——更精确但实现更复杂

为什么工厂返回值而不是工具列表？

优势：灵活——可以返回任何类型（工具列表、单个工具、None） 代价：返回值类型不确定，需要运行时检查 替代方案：强制返回列表——类型安全但灵活性差

参考来源

• Agno agent/_run.py resolve_run_dependencies
• Agno agent/agent.py dependencies 字段