大模型Agent智能体开发实战：工具调用编排与多步推理工作流设计

小编 • • 人工智能

大模型Agent智能体正在从简单的问答对话走向复杂的任务执行系统。Agent的核心能力在于将大语言模型的推理能力与外部工具调用相结合，通过多步推理完成单次对话无法解决的复杂任务。本文围绕Agent智能体开发中的工具调用协议、ReAct推理框架、多步工作流编排等关键技术展开，提供可直接复用的代码实现。

Agent智能体架构与核心组件

Agent智能体的本质是一个自主决策循环：感知输入、规划行动、执行工具、观察结果、更新状态，直到任务完成或达到终止条件。与传统的Chain-of-LangChain链式调用不同，Agent的执行路径不是预定义的，而是由模型在运行时动态决定。

一个完整的Agent系统包含以下组件：

– LLM核心：负责推理和决策，支持function calling或tool use接口
– 工具集：搜索、计算、数据库查询、API调用等可执行函数
– 记忆模块：短期对话上下文与长期向量记忆
– 规划器：将用户目标分解为可执行的子任务序列
– 执行引擎：调度工具调用，处理异常与重试

Function Calling工具定义规范

OpenAI、Anthropic、Google等厂商的function calling接口格式略有差异，但核心思路一致：将工具能力以JSON Schema形式声明，模型根据上下文选择调用哪个工具、传入什么参数。

from openai import OpenAI
import json

client = OpenAI()

# 定义工具集
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "在产品数据库中搜索匹配条件的商品",
            "parameters": {
                "type": "object",
                "properties": {
                    "keyword": {"type": "string", "description": "搜索关键词"},
                    "max_price": {"type": "number", "description": "最高价格"},
                    "category": {"type": "string", "enum": ["电子", "服装", "食品"]}
                },
                "required": ["keyword"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_discount",
            "description": "根据商品价格和折扣规则计算最终价格",
            "parameters": {
                "type": "object",
                "properties": {
                    "original_price": {"type": "number", "description": "原价"},
                    "discount_rate": {"type": "number", "description": "折扣率，0.8表示8折"}
                },
                "required": ["original_price", "discount_rate"]
            }
        }
    }
]

# 工具实现
def search_database(keyword, max_price=None, category=None):
    results = [
        {"name": "无线蓝牙耳机", "price": 199, "category": "电子"},
        {"name": "智能手表", "price": 899, "category": "电子"},
    ]
    filtered = [r for r in results if keyword.lower() in r["name"].lower()]
    if max_price:
        filtered = [r for r in filtered if r["price"] <= max_price]
    return json.dumps(filtered, ensure_ascii=False)

def calculate_discount(original_price, discount_rate):
    final_price = round(original_price * discount_rate, 2)
    return json.dumps({"original": original_price, "final": final_price}, ensure_ascii=False)

tool_map = {
    "search_database": search_database,
    "calculate_discount": calculate_discount
}

ReAct推理框架实现多步工具调用

ReAct（Reasoning + Acting）框架是Agent智能体最经典的推理范式。模型在每一步先输出思考过程（Thought），再决定行动（Action），执行后观察结果（Observation），循环往复直到得出最终答案。

def run_agent(user_query, max_steps=10):
    messages = [
        {"role": "system", "content": "你是一个购物助手Agent，可以搜索商品和计算折扣。请逐步思考并使用工具完成任务。"},
        {"role": "user", "content": user_query}
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        msg = response.choices[0].message
        messages.append(msg)

        if msg.tool_calls:
            for tc in msg.tool_calls:
                func_name = tc.function.name
                func_args = json.loads(tc.function.arguments)
                print(f"[Step {step+1}] 调用工具: {func_name}({func_args})")

                result = tool_map[func_name](**func_args)
                print(f"[Step {step+1}] 工具返回: {result}")

                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result
                })
        else:
            print(f"[Step {step+1}] Agent最终回答: {msg.content}")
            return msg.content

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

answer = run_agent("帮我找一款200元以内的蓝牙耳机，然后算一下打8折多少钱")

多Agent协作与工作流编排

面对复杂任务，单Agent的能力存在上限。多Agent系统通过角色分工和消息传递实现协作：一个Coordinator Agent负责任务拆解和分发，多个Worker Agent各司其职，结果汇总后由Coordinator给出最终输出。

from enum import Enum
from dataclasses import dataclass

class AgentRole(Enum):
    COORDINATOR = "coordinator"
    RESEARCHER = "researcher"
    CODER = "coder"
    REVIEWER = "reviewer"

@dataclass
class TaskMessage:
    sender: str
    receiver: str
    content: str
    task_id: str
    status: str  # pending, in_progress, completed, failed

class MultiAgentOrchestrator:
    def __init__(self):
        self.agents = {}
        self.message_queue = []

    def register_agent(self, name, role, system_prompt):
        self.agents[name] = {
            "role": role,
            "system_prompt": system_prompt,
            "history": []
        }

    def dispatch_task(self, task_description):
        sub_tasks = self._decompose_task(task_description)
        results = []

        for sub_task in sub_tasks:
            agent_name = self._select_agent(sub_task["type"])
            result = self._execute_agent(agent_name, sub_task["prompt"])
            results.append({"agent": agent_name, "result": result})

        final = self._synthesize(results, task_description)
        return final

    def _decompose_task(self, task):
        return [
            {"type": "research", "prompt": f"调研: {task}"},
            {"type": "coding", "prompt": f"实现: {task}"},
            {"type": "review", "prompt": f"审查: {task}"}
        ]

    def _select_agent(self, task_type):
        mapping = {"research": "researcher", "coding": "coder", "review": "reviewer"}
        return mapping.get(task_type, "coordinator")

    def _execute_agent(self, agent_name, prompt):
        agent = self.agents[agent_name]
        messages = [{"role": "system", "content": agent["system_prompt"]}]
        messages.append({"role": "user", "content": prompt})
        resp = client.chat.completions.create(model="gpt-4o", messages=messages)
        return resp.choices[0].message.content

    def _synthesize(self, results, original_task):
        summary = "\n".join([f"[{r['agent']}] {r['result'][:200]}..." for r in results])
        return f"任务: {original_task}\n\n协作结果:\n{summary}"

Agent记忆与状态持久化

长期运行的Agent需要跨越单次会话的记忆能力。短期记忆用于当前对话上下文，长期记忆通过向量数据库存储历史交互，在需要时检索相关片段注入上下文。

import chromadb

class AgentMemory:
    def __init__(self, collection_name="agent_memory"):
        self.client = chromadb.PersistentClient(path="./agent_memory_db")
        self.collection = self.client.get_or_create_collection(name=collection_name)

    def store(self, text, metadata=None, doc_id=None):
        self.collection.add(
            documents=[text],
            metadatas=[metadata or {}],
            ids=[doc_id or f"mem_{self.collection.count()}"]
        )

    def retrieve(self, query, top_k=5):
        results = self.collection.query(query_texts=[query], n_results=top_k)
        return results["documents"][0] if results["documents"] else []

memory = AgentMemory()
memory.store("用户偏好购买性价比高的电子产品", {"type": "preference"})
memory.store("用户上次购买了蓝牙耳机", {"type": "history"})

relevant = memory.retrieve("用户想买什么电子产品")
context = "\n".join(relevant)

Agent开发中的常见陷阱与解决方案

工具调用幻觉：模型编造不存在的工具或传入错误参数。解决方案是在工具描述中明确约束，并在执行前做参数校验。

无限循环：Agent反复调用同一工具无法收敛。设置max_steps硬限制，并在prompt中要求模型在工具调用失败后调整策略而非重试相同操作。

上下文窗口溢出：多轮工具调用的历史消息快速膨胀。使用滑动窗口或摘要压缩策略，保留最近的N轮交互，将更早的历史压缩为摘要。

def compress_context(messages, keep_recent=10):
    if len(messages) <= keep_recent:
        return messages

    system_msgs = [m for m in messages if m["role"] == "system"]
    recent = messages[-(keep_recent - len(system_msgs)):]
    early = messages[len(system_msgs):-len(recent)]
    summary = compress_with_llm(early)

    return system_msgs + [
        {"role": "system", "content": f"历史摘要: {summary}"}
    ] + recent

Agent评估与可观测性

Agent系统调试难度远高于普通LLM应用，因为执行路径不确定。需要记录每一步的思考、工具调用、返回结果，形成完整的trace日志。

import logging
from datetime import datetime

class AgentTracer:
    def __init__(self, agent_name):
        self.logger = logging.getLogger(agent_name)
        self.logger.setLevel(logging.DEBUG)
        handler = logging.FileHandler(f"agent_trace_{agent_name}.log", encoding="utf-8")
        handler.setFormatter(logging.Formatter("%(asctime)s | %(message)s"))
        self.logger.addHandler(handler)

    def trace_step(self, step, thought, action, observation, duration_ms):
        self.logger.info(
            f"Step {step} | Thought: {thought[:100]}... | "
            f"Action: {action} | Duration: {duration_ms}ms | "
            f"Observation: {str(observation)[:200]}..."
        )

    def trace_error(self, step, error):
        self.logger.error(f"Step {step} | ERROR: {error}")

Agent智能体的工程化落地需要在推理能力、工具编排、状态管理和可观测性之间找到平衡。从单Agent的ReAct循环到多Agent协作编排，核心是将LLM的推理能力转化为可执行的任务流，同时通过完善的trace机制确保系统行为可追溯、可调试。

原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/da-mo-xing-agent-zhi-neng-ti-kai-fa-shi-zhan-gong-ju-diao/

赞 (0)
小编小编
0
上一篇 1小时前
下一篇 1小时前

相关推荐