LangChain实战：从零构建Agent Skill的核心技术与避坑指南

5次阅读

共计 3272 个字符，预计需要花费 9 分钟才能阅读完成。

传统对话系统（如基于规则或简单意图识别）在处理多步骤复杂任务时常常力不从心。比如用户说 ” 帮我订机票然后预订机场附近的酒店 ”，这类需要多个子任务协同的场景会暴露三个典型问题：

上下文断裂：每个子任务被当作独立请求处理
缺乏工具协调：无法动态调用不同功能模块（如机票 API 和酒店 API）
状态管理困难：用户中途修改需求时（如 ” 先不要机票了 ”）难以回滚

Agent 架构通过引入任务分解（Task Decomposition）、工具链（Tool Chain）和记忆管理（Memory）三大核心组件，让系统具备了类似人类的 ” 思考 - 行动 - 观察 ” 循环能力。

与 Rasa、DialogFlow 等传统框架对比，LangChain 在开发复杂 Agent 时展现明显优势：

特性	LangChain	Rasa	DialogFlow
动态工具调用	✅	❌	❌
长期记忆	✅	有限	有限
开源可定制性	✅	✅	❌
学习曲线	中等	陡峭	平缓

特别在需要结合大语言模型（LLM）的场景下，LangChain 的 AgentExecutor 提供了开箱即用的任务调度能力。

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.memory import ConversationBufferWindowMemory
from langchain_community.tools import DuckDuckGoSearchResults

# 工具配置（Tool Chain）tools = [DuckDuckGoSearchResults(name="web_search")]  

# 记忆管理（Memory）memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    k=5,  # 保留最近 5 轮对话
    return_messages=True
)

# LLM 配置（建议生产环境使用 Azure OpenAI 等商用方案）from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 完整 Agent 组装
agent = create_openai_tools_agent(llm, tools, "你是一个旅行助手")
agent_executor = AgentExecutor(agent=agent, tools=tools, memory=memory)

当用户输入复杂请求时，需要先拆解为子任务。以下是带错误重试的版本：

from langchain_core.runnables import RunnableLambda
import random

def task_decomposition(user_input: str) -> list[str]:
    """将复杂任务拆解为子任务列表"""
    retry_count = 0
    max_retries = 3

    while retry_count < max_retries:
        try:
            # 实际生产应使用更可靠的分解策略
            if "然后" in user_input:
                return user_input.split("然后")
            return [user_input]
        except Exception as e:
            retry_count += 1
            wait_time = random.uniform(0.5, 2.0)
            time.sleep(wait_time)  # 随机退避避免雪崩

    raise ValueError(f"任务分解失败，原始输入：{user_input}")

decomposition_chain = RunnableLambda(task_decomposition)

sequenceDiagram
    participant User
    participant Agent
    participant Tool1
    participant Tool2

    User->>Agent: 复杂请求（如 "查天气然后推荐穿搭"）Agent->>decomposition_chain: 拆解任务
    decomposition_chain-->>Agent: ["查天气", "推荐穿搭"]

    loop 每个子任务
        Agent->>Tool1: 执行子任务 1（查天气）Tool1-->>Agent: 天气数据
        Agent->>Memory: 保存中间结果

        Agent->>Tool2: 执行子任务 2（推荐穿搭）Tool2-->>Agent: 穿搭建议
    end

    Agent->>User: 整合最终响应

# 更精细化的记忆控制
from langchain.memory import (
    ConversationBufferWindowMemory,
    CombinedMemory
)

# 组合式记忆：短期记忆 + 重要事件记忆
short_memory = ConversationBufferWindowMemory(k=5)
important_memory = ConversationBufferWindowMemory(
    memory_key="important_facts",
    k=10,
    input_key="only_important"  # 需在前置步骤过滤关键信息
)

memory = CombinedMemory(memories=[short_memory, important_memory])

from threading import Lock

tool_lock = Lock()

class SafeTool:
    def __init__(self, tool):
        self.tool = tool

    def run(self, input_str):
        with tool_lock:  # 避免多个 Agent 同时调用
            return self.tool.run(input_str)

# 使用装饰器更优雅（生产推荐）from functools import wraps

def concurrency_lock(func):
    lock = Lock()

    @wraps(func)
    def wrapper(*args, **kwargs):
        with lock:
            return func(*args, **kwargs)
    return wrapper

工具幂等性：
问题：网络工具被重复调用可能导致重复订票等严重错误

解决：为工具添加唯一请求 ID（UUID）和结果缓存

from uuid import uuid4

def booking_tool(destination: str):
    request_id = f"{destination}_{uuid4()}"
    if cache.get(request_id):
        return cache[request_id]
    # 真实业务逻辑...

记忆窗口太小：
问题：当 k = 3 时，长对话中早期关键信息会丢失
解决：采用分层记忆，关键事实永久存储
异步超时失控：
问题：工具调用卡死导致整个 Agent 无响应

解决：为所有工具添加超时限制

from concurrent.futures import TimeoutError

async def safe_tool_call(tool, input_str, timeout=10):
    try:
        return await asyncio.wait_for(tool.arun(input_str), 
            timeout=timeout
        )
    except TimeoutError:
        return "该功能暂时不可用"