LangChain与AI Agent架构设计：从基础原理到生产环境实践

1次阅读

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

在直接使用大模型 API 开发对话系统时，开发者常遇到以下典型问题：

状态管理混乱：对话历史、用户上下文等状态信息常与业务逻辑耦合
代码重复率高：相似的前后处理逻辑（如 prompt 模板、结果解析）需要重复实现
扩展成本高：新增功能（如工具调用、多轮对话）时需重构核心流程
监控困难：缺乏标准化的日志和异常处理机制

对比维度	原生 API 方案	LangChain 方案
维护性	需自行管理 prompt 模板	内置 PromptTemplate 标准化管理
扩展性	功能迭代需修改主流程	通过 Chain 组件插拔式扩展
状态管理	上下文需手动拼接	Memory 组件自动维护对话历史
错误处理	需自定义重试机制	内置 retry 和 fallback 策略

Agent
– 决策中枢，根据输入选择执行哪个 Chain
– 典型实现方式：ZeroShotAgent、ReActAgent

Chain
– 执行单元，组合 LLM 调用与其他操作
– 基础类型：LLMChain、TransformChain
– 组合类型：SequentialChain、RouterChain

Memory
– 状态存储器，维护对话上下文
– 常见实现：ConversationBufferMemory、RedisChatMessageHistory

from langchain.chains import LLMChain
from langchain.memory import ConversationBufferMemory
from langchain.prompts import PromptTemplate
from langchain_community.llms import OpenAI

# 定义带有历史记录的 prompt 模板
dialog_template = """
你是一个客服助手，根据对话历史回答用户问题：历史记录：{history}

当前输入：{input}
"""

prompt = PromptTemplate(input_variables=["history", "input"],
    template=dialog_template
)

# 初始化带记忆的 chain
memory = ConversationBufferMemory(memory_key="history")
dialog_chain = LLMChain(llm=OpenAI(temperature=0.7),
    prompt=prompt,
    memory=memory,
    verbose=True  # 开启调试日志
)

# 带异常处理的对话循环
while True:
    try:
        user_input = input("用户:")
        if user_input.lower() == 'exit':
            break

        response = dialog_chain.run(input=user_input)
        print(f"助手: {response}")

    except Exception as e:
        print(f"[ERROR] 对话异常: {str(e)}")
        # 可加入重试逻辑或 fallback 响应

# 查看当前记忆内容
print(memory.buffer)

# 保存对话历史到文件
import json
with open('dialog_history.json', 'w') as f:
    json.dump(memory.load_memory_variables({}), f)

# 从文件恢复对话状态
with open('dialog_history.json') as f:
    history_data = json.load(f)
    memory.save_context({"input": history_data["history"][::2]},  # 用户输入
        {"output": history_data["history"][1::2]} # 助手回复
    )

速率控制

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(chain, input_text):
    return chain.run(input=input_text)

异步处理

from langchain.chains import LLMChain
from langchain_community.llms import OpenAI

async def process_concurrent_requests(requests):
    semaphore = asyncio.Semaphore(5)  # 并发控制
    async with semaphore:
        return await chain.arun(input=requests)

方案	适用场景	示例实现
Redis	高频短对话	`RedisChatMessageHistory`
PostgreSQL	审计需求强	`PostgresChatMessageHistory`
MongoDB	非结构化上下文	`MongoDBChatMessageHistory`

3 层嵌套规则：任何 Chain 的嵌套不超过 3 层
单一职责：每个 Chain 只处理一类任务
明确接口：定义清晰的 input/output 变量名

使用 tracemalloc 监控

import tracemalloc

tracemalloc.start()
# 执行 Chain 操作
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
print("[Memory] Top allocations:")
for stat in top_stats[:5]:
    print(stat)

定期清理 Memory

# 每 20 轮对话清理一次历史
if len(memory.buffer) > 20:
    memory.clear()

TODO：实现天气查询工具集成

创建 Tool 组件

from langchain.tools import BaseTool

class WeatherTool(BaseTool):
    name = "weather_query"
    description = "查询城市天气情况"

    def _run(self, city: str) -> str:
        # 调用天气 API
        return f"{city}天气：晴，25℃"

修改 Agent 初始化

from langchain.agents import initialize_agent

agent = initialize_agent(tools=[WeatherTool()],
    llm=OpenAI(temperature=0),
    agent="zero-shot-react-description",
    verbose=True
)

测试工具调用

response = agent.run("上海现在天气如何？")
print(response)

对于生产级 AI Agent 系统，建议采用渐进式架构演进：
1. 初期先用 LLMChain+ConversationBufferMemory 验证核心流程
2. 中期引入 Agent+Tools 实现功能扩展
3. 后期通过 Custom Memory 和Async Support优化性能

关键监控指标应包括：Chain 执行耗时、Memory 增长速率、API 调用错误率等。建议使用 LangSmith 等可视化工具进行链路追踪。

正文完