LangChain与AI Agent实战入门：从零构建智能对话系统的避坑指南

1次阅读

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

传统对话系统开发中，开发者常遇到两个核心问题：

上下文断裂：普通 API 调用无法自动维护多轮对话状态，需要自行实现会话 ID 匹配和上下文拼接逻辑，代码复杂度高
工具调度僵化：当需要调用外部工具（如数据库查询、API 请求）时，传统方案往往需要硬编码判断条件，扩展性和可维护性差

我曾维护过一个基于 Flask 的客服系统，每次新增业务接口都要修改路由逻辑，且上下文管理代码占总量的 40% 以上。

维度	直接调用 API	LangChain 方案
上下文维护	需自行实现	内置 Memory 模块自动处理
工具调用	人工编排调用顺序	动态路由决策
延迟(平均)	300-500ms（简单请求）	增加 150-200ms（决策开销）
成本	按 token 计费	相同计费模式 + 少量内存消耗
开发效率	低（需处理底层细节）	高（声明式编程）

测试环境：AWS g4dn.xlarge + GPT-3.5-turbo，10 次请求平均值

# 建议 Python 3.10+
pip install langchain openai python-dotenv

from langchain.agents import AgentExecutor, Tool
from langchain.agents.openai_functions_agent import OpenAIFunctionsAgent
from langchain.memory import ConversationBufferMemory
from langchain.chat_models import ChatOpenAI
from langchain.prompts import MessagesPlaceholder

# 示例工具定义
def search_order(order_id: str) -> str:
    """模拟订单查询工具"""
    return f"订单 {order_id} 状态：已发货"

# 关键组件初始化
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo")

tools = [
    Tool(
        name="OrderSearch",
        func=search_order,
        description="根据订单 ID 查询物流状态"
    )
]

# 特别注意：这是 LangChain 0.0.200+ 的新 API 格式
agent = OpenAIFunctionsAgent.from_llm_and_tools(
    llm=llm,
    tools=tools,
    extra_prompt_messages=[MessagesPlaceholder(variable_name="chat_history")]
)

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True  # 调试时建议开启
)

# 执行示例
result = executor.run("帮我查下订单 12345 的状态")
print(result)

关键点说明：

ConversationBufferMemory 会以列表形式保存对话历史
Tool 定义时必须包含清晰的 description，这是 LLM 决策的依据
verbose=True 时会打印完整的 Chain 执行过程

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
)

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def safe_llm_call(prompt: str) -> str:
    """带指数退避的重试机制"""
    return llm(prompt)

from langchain.memory import RedisChatMessageHistory

message_history = RedisChatMessageHistory(
    session_id="user123",
    url="redis://localhost:6379/0",
    ttl=600  # 10 分钟过期
)

memory = ConversationBufferMemory(
    memory_key="chat_history",
    chat_memory=message_history,
    return_messages=True
)

输入校验：对用户输入进行正则过滤（如移除 <>{} 等特殊字符）
权限隔离：不同功能的 Tool 应设置独立的 API 访问权限
监控：记录 LLM 的完整输出日志，定期审计

# 获取 token 级概率输出
from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = executor.run("测试问题")
    print(f"Used tokens: {cb.total_tokens}")
    print(f"Top alternatives: {cb.response_msgs[0].logprobs}")