F FisherHub Docs

Agent 开发

什么是 LLM Agent

Agent(智能体)是指能够感知环境、自主规划和执行行动以达成目标的系统。在大模型语境下,Agent 结合了 LLM 的推理能力和外部工具的调用能力,可以完成复杂的多步骤任务。

Agent 的核心循环:

用户请求 → LLM 理解意图 → 规划步骤 → 调用工具 → 观察结果 → 调整计划 → 输出最终结果

Function Calling

Function Calling 是 LLM Agent 的基础设施。它让模型输出结构化的函数调用参数,由应用层实际执行函数并返回结果。

import json
from openai import OpenAI

client = OpenAI()

# 定义可用的工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称,如北京、上海"
                    },
                    "date": {
                        "type": "string",
                        "description": "日期,格式 YYYY-MM-DD"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_news",
            "description": "搜索最新新闻",
            "parameters": {
                "type": "object",
                "properties": {
                    "keyword": {"type": "string"},
                    "limit": {"type": "integer", "default": 5}
                },
                "required": ["keyword"]
            }
        }
    }
]

# 执行函数调用
def call_llm_with_tools(messages):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )
    return response.choices[0].message

# 处理工具调用
def execute_tool_call(tool_call):
    fn_name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)

    if fn_name == "get_weather":
        return get_weather(**args)
    elif fn_name == "search_news":
        return search_news(**args)
    else:
        raise ValueError(f"未知工具:{fn_name}")

ReAct 模式实现

ReAct(Reasoning + Acting)让模型交替进行推理和行动,每一步都记录思考过程和观察结果。

def agent_loop(user_request, max_steps=10):
    messages = [{"role": "user", "content": user_request}]

    for step in range(max_steps):
        response = call_llm_with_tools(messages)
        messages.append(response)

        if not response.tool_calls:
            # 模型没有调用工具,直接返回最终回答
            return response.content

        for tool_call in response.tool_calls:
            result = execute_tool_call(tool_call)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })

    return "达到最大步数限制,任务可能未完成。"

多 Agent 协作框架

单个 Agent 在处理复杂任务时容易陷入局部最优。多 Agent 系统通过分工协作来提升效果。

1. 主管-员工模式(Orchestrator-Worker)

一个主管 Agent 负责任务拆分和调度,多个员工 Agent 并行执行子任务。

class OrchestratorAgent:
    def __init__(self):
        self.workers = {
            "research": ResearchAgent(),
            "writer": WriterAgent(),
            "reviewer": ReviewerAgent()
        }

    def run(self, task):
        plan = self.decompose_task(task)
        for step in plan:
            worker = self.workers[step.agent_type]
            result = worker.execute(step.description)
            self.track_progress(step, result)
        return self.synthesize_results()

2. 辩论模式(Debate)

多个 Agent 就同一问题各自生成答案,然后互相评审和辩论,最终达成共识。

问题:这个设计方案的优缺点是什么?

Agent A(架构师视角):分析可扩展性和系统耦合
Agent B(运维视角):分析部署成本和可靠性
Agent C(安全视角):分析潜在安全风险

→ 三方交叉评审 → 综合报告

3. 流水线模式(Pipeline)

将任务拆成链式处理的多个阶段,每个 Agent 专注于一个环节。

需求分析 Agent → 架构设计 Agent → 代码生成 Agent → 代码审查 Agent → 测试 Agent

Agent 工具设计原则

工具的粒度

工具应该职责单一。一个工具只做一件事,做清楚。

❌ 一个万能工具:run_task(task_type, params)
✅ 多个专用工具:create_file, search_db, send_email, calculate_price

工具的描述

充分描述工具的用途和使用条件,帮助 LLM 在需要时主动选择。

tools = [
    {
        "function": {
            "name": "run_sql_query",
            "description": "对生产数据库执行只读 SQL 查询。"
                          "仅支持 SELECT 语句,不可执行 INSERT/UPDATE/DELETE。"
                          "复杂查询请先说明执行计划。",
            ...
        }
    }
]

错误处理

工具调用可能失败,Agent 需要能从错误中恢复。

def safe_execute_tool(tool_call):
    try:
        return execute_tool_call(tool_call)
    except RateLimitError:
        return {"error": "rate_limited", "retry_after": 60}
    except AuthError:
        return {"error": "unauthorized", "advice": "请先登录"}
    except Exception as e:
        return {"error": str(e), "can_retry": True}

当前局限与应对

  1. 幻觉循环:Agent 可能反复执行错误操作,需要设置步数上限和兜底逻辑
  2. 上下文窗口:多步 Agent 会积累大量对话内容,需要记忆压缩或滑动窗口
  3. 工具选择:当工具数量超过 30 个时,模型的选择准确率会下降,需要分级路由