基于LangChain实现多Agent协作系统的开发案例

# 基于LangChain实现多Agent协作系统的开发案例

## 引言
在大语言模型应用开发中，多智能体（Multi-Agent）协作系统展现出强大的任务处理能力。本文将通过一个电商客服场景案例（订单查询+退换货处理），演示如何使用LangChain框架构建多Agent协作系统。

---

## 准备工作

### 环境要求
- Python 3.8+
- LangChain 0.1.0+
- OpenAI API Key

# 安装核心依赖
pip install langchain langchain-openai python-dotenv

OPENAI_API_KEY=your_api_key_here

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

class BaseAgent:
    def __init__(self, name, tools, system_prompt):
        self.name = name
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

        prompt = ChatPromptTemplate.from_messages([
            ("system", system_prompt),
            MessagesPlaceholder("chat_history", optional=True),
            ("human", "{input}"),
            MessagesPlaceholder("agent_scratchpad")
        ])

        self.agent = create_openai_tools_agent(self.llm, tools, prompt)
        self.executor = AgentExecutor(agent=self.agent, tools=tools)

from langchain.chains import ConversationChain

class MultiAgentSystem:
    def __init__(self):
        # 初始化对话记忆链
        self.memory = ConversationChain(
            llm=ChatOpenAI(temperature=0),
            memory=ConversationBufferMemory()
        )

        # 创建不同职能的Agent实例
        self.order_agent = BaseAgent(
            name="OrderExpert",
            tools=[order_tool],
            system_prompt="您是订单专家..."
        )

        self.return_agent = BaseAgent(
            name="ReturnSpecialist",
            tools=[return_tool],
            system_prompt="您是退换货专家..."
        )

    def route_query(self, user_input):
        # 记忆上下文管理（关键！）
        context = self.memory.predict(input=user_input)

        # 基于上下文的路由逻辑示例
        if "订单状态" in context:
            return self.order_agent.executor.invoke({"input": context})
        elif "退货" in context or "换货" in context:
            return self.return_agent.executor.invoke({"input": context})

from langchain.tools import tool

@tool
def order_tool(order_id: str) -> str:
    """根据订单ID查询订单状态"""
    # Mock数据库查询逻辑示例 
    orders = {
        "1001": {"status": "已发货", "items": ["商品A"]},
        "1002": {"status": "处理中", "items": ["商品B"]}
    }
    return f"订单状态：{orders.get(order_id, '未找到')}"

# multi_agent_system.py（完整实现）
import os 
from dotenv import load_dotenv

load_dotenv()

# [此处插入上述BaseAgent和MultiAgentSystem的实现]

if __name__ == "__main__":
    system = MultiAgentSystem()

    while True:
        query = input("用户提问：")

        # Step1: 路由决策 & Step2: Agent执行 
        response = system.route_query(query)

        # Step3: 更新对话记忆 
        system.memory.save_context(
            {"input": query}, 
            {"output": response["output"]}
        )

        print(f"系统回复：{response['output']}")

# Agent执行器中添加中断机制 
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=3,   # ❗️限制最大调用次数 
    early_stopping_method="generate"
)