LangChain与AI Agent开发踩坑记录

LangChain去年火了一段时间，我折腾了几个月，把踩的坑和实战经验记录一下。核心就是Agent、Chain、Memory这几个概念，上手不难，但坑不少。

环境准备

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

# 向量存储（可选）
pip install pinecone-client==2.2.1

环境变量配置：

# .env文件
OPENAI_API_KEY=your_api_key
PINECONE_API_KEY=your_pinecone_key

LLM基础调用

LangChain支持多种模型，OpenAI的最稳定。

from langchain.schema import HumanMessage
from langchain.chat_models import ChatOpenAI

# 初始化模型
llm = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    temperature=0,
    openai_api_key="your_api_key"
)

# 基础调用
output = llm([HumanMessage(content="Translate to French: I love programming.")])
print(output)  # AIMessage(content="J'aime programmer.")

带角色设定的对话：

from langchain.schema import SystemMessage

messages = [
    SystemMessage(content="You are a helpful assistant that translates English to French."),
    HumanMessage(content="I love programming.")
]
output = llm(messages)

批量生成：

batch_messages = [
    [SystemMessage(content="Translate to French"), HumanMessage(content="I love programming.")],
    [SystemMessage(content="Translate to French"), HumanMessage(content="Hello world.")],
]
output = llm.generate(batch_messages)

提示模板Prompt Templates

Prompt模板让代码更灵活：

from langchain.prompts.chat import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
)

# 定义系统模板
system_template = "You are a translator from {input_language} to {output_language}."
system_message = SystemMessagePromptTemplate.from_template(system_template)

# 用户模板
human_template = "{text}"
human_message = HumanMessagePromptTemplate.from_template(human_template)

# 组合
chat_prompt = ChatPromptTemplate.from_messages([system_message, human_message])

# 使用
prompt = chat_prompt.format_prompt(
    input_language="English",
    output_language="French",
    text="I love programming."
).to_messages()

output = llm(prompt)

Chain链式调用

Chain把多个步骤串起来。

基础LLMChain：

from langchain import LLMChain

chain = LLMChain(llm=llm, prompt=prompt)
print(chain.run("autoencoder"))

顺序链SequentialChain，实现多步骤流程：

from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import ChatPromptTemplate

# 步骤1：翻译评论
first_prompt = ChatPromptTemplate.from_template(
    "Translate the following review to english:\n\n{Review}"
)
chain_one = LLMChain(
    llm=llm,
    prompt=first_prompt,
    output_key="English_Review"
)

# 步骤2：一句话总结
second_prompt = ChatPromptTemplate.from_template(
    "Summarize the following review in 1 sentence:\n\n{English_Review}"
)
chain_two = LLMChain(
    llm=llm,
    prompt=second_prompt,
    output_key="summary"
)

# 步骤3：检测语言
third_prompt = ChatPromptTemplate.from_template(
    "What language is this review? Only answer the language name:\n\n{Review}"
)
chain_three = LLMChain(
    llm=llm,
    prompt=third_prompt,
    output_key="language"
)

# 步骤4：生成回复
fourth_prompt = ChatPromptTemplate.from_template(
    "Write a follow up response to this summary in the specified language:\n\n"
    "Summary: {summary}\n\nLanguage: {language}"
)
chain_four = LLMChain(
    llm=llm,
    prompt=fourth_prompt,
    output_key="followup_message"
)

# 组合顺序链
overall_chain = SequentialChain(
    chains=[chain_one, chain_two, chain_three, chain_four],
    input_variables=["Review"],
    output_variables=["English_Review", "summary", "followup_message"],
    verbose=True
)

# 执行
review = "Je trouve le goût médiocre. La mousse ne tient pas."
result = overall_chain(review)

Agent智能体

Agent是LangChain的核心，让LLM能调用工具完成任务。

ReAct框架

ReAct（Reasoning and Acting）的核心循环：

Thought → Action → Observation → Thought → ...
思考    → 执行    → 观察结果   → 再思考

Agent自己决定什么时候用工具，什么时候输出答案。

基础Agent实现

from langchain.agents import load_tools, initialize_agent, AgentType

# 加载工具（llm-math是做数学计算的）
tools = load_tools(["llm-math"], llm=llm)

# 初始化Agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 运行
try:
    output = agent.run(
        "What was the high temperature in SF yesterday in Fahrenheit? "
        "What is that number raised to the .023 power?"
    )
    print(output)
except Exception as e:
    print(f"Error: {e}")

执行过程大概这样：

> Entering new AgentExecutor chain...
Thought: I need to find the temperature and calculate...
Action: Use calculator
Action Input: 72.21 ^ 0.023
Observation: 1.023...
Thought: I now know the final answer
Final Answer: 1.023...
> Finished chain.

结构化对话Agent

复杂工具需要结构化参数：

from langchain.tools import StructuredTool

def ssh(command: str, host: str, username: str = "root") -> str:
    """Connect to remote server and execute commands."""
    import os
    return os.popen(f"ssh {host} -l{username} '{command}'").read()

ssh_tool = StructuredTool.from_function(ssh)

# 初始化结构化Agent
agent = initialize_agent(
    [ssh_tool],
    OpenAI(temperature=0),
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 执行
result = agent.run("帮我看一下 8.130.35.2 这台机器运行多久了")

Agent会自动解析参数并调用：

Action: ssh
{
  "action": "ssh",
  "action_input": {
    "command": "uptime",
    "host": "8.130.35.2"
  }
}

Observation: 15:48:44 up 25 days, 41 min, load average: 1.04
Thought: I have the answer
Final Answer: This machine has been running for 25 days and 41 minutes.

自定义Python执行Agent

写了个可以执行Python代码的Agent：

from typing import Dict
import traceback
from io import StringIO
from contextlib import redirect_stdout, redirect_stderr

def python_executor(code_str: str, return_context: bool = False) -> Dict:
    """Execute Python code and return output."""
    stdout = StringIO()
    stderr = StringIO()
    context = {}

    try:
        with redirect_stdout(stdout), redirect_stderr(stderr):
            exec(code_str, context)
    except Exception:
        stderr.write(traceback.format_exc())

    if return_context:
        return {
            "stdout": stdout.getvalue()[:1000],
            "stderr": stderr.getvalue()[:1000],
            "context": context
        }
    else:
        return {
            "stdout": stdout.getvalue()[:1000],
            "stderr": stderr.getvalue()[:1000]
        }

# 创建Agent
agent = initialize_agent(
    [StructuredTool.from_function(python_executor)],
    OpenAI(temperature=0.3),
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 让它抓取RSS
resp = agent.run(
    "帮我利用工具总结一下 https://example.com/blog/feed 这个xml的博客订阅"
)

Memory记忆机制

对话需要上下文记忆，LangChain内置了几种Memory。

基础ConversationChain：

from langchain import ConversationChain

# 自带记忆
conversation = ConversationChain(
    llm=llm,
    verbose=True
)

# 多轮对话
output1 = conversation.predict(input="Hi there!")
print(output1)  # "Hello! How are you today?"

output2 = conversation.predict(input="I'm doing well!")
print(output2)  # 记住上文，给出连贯回复

消息角色系统：

def get_completion_from_messages(messages, model="gpt-3.5-turbo", temperature=0):
    """基于消息历史的对话"""
    response = openai.ChatCompletion.create(
        model=model,
        messages=messages,
        temperature=temperature,
    )
    return response.choices[0].message["content"]

# 构建对话历史
messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "Hi there!"},
    {"role": "assistant", "content": "Hello! How can I help you?"},
    {"role": "user", "content": "What's the weather like?"}
]

response = get_completion_from_messages(messages)

角色说明：

• system：定义AI行为准则
• user：用户输入
• assistant：AI回复

实战案例：AI客服机器人

写了个简单的客服机器人：

from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain

# 客服提示模板
customer_service_prompt = ChatPromptTemplate.from_template("""
You are a customer service AI assistant.
Your task is to send an email reply to a valued customer.

Given the customer email delimited by ```,
Generate a reply to thank the customer for their review.

If the sentiment is positive or neutral, thank them for their review.
If the sentiment is negative, apologize and suggest that they can reach out to customer service.

Make sure to use specific details from the review.
Write in a concise and professional tone.
Sign the email as `AI customer agent`.

Customer review: ```{review}```
Review sentiment: {sentiment}
""")

# 创建客服链
service_chain = LLMChain(
    llm=llm,
    prompt=customer_service_prompt,
    verbose=True
)

# 情感分析函数
def analyze_sentiment(review: str) -> str:
    sentiment_prompt = f"Classify the sentiment of this review: {review}"
    result = llm([HumanMessage(content=sentiment_prompt)])
    return result.content.lower()

# 完整流程
def customer_service_reply(review: str) -> str:
    sentiment = analyze_sentiment(review)
    reply = service_chain.run(review=review, sentiment=sentiment)
    return reply

# 测试
review = "The product is amazing! Fast delivery and great quality."
reply = customer_service_reply(review)
print(reply)

踩过的坑

OutputParser错误

经常遇到这个报错：

OutputParserException: Parsing LLM output produced both
a final answer and a parse-able action

原因是Agent同时输出了Final Answer和Action，Parser处理不了。

修复方案，改output_parser.py：

# 原代码
if action_match:
    if includes_answer:
        raise OutputParserException(...)

# 修复后
if action_match and not includes_answer:
    if includes_answer:
        raise OutputParserException(...)

其实就是加了个条件判断，避免重复处理。

调试技巧

开verbose模式看Agent的思考过程：

import logging
logging.basicConfig(level=logging.DEBUG)

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,  # 详细输出
    return_intermediate_steps=True  # 返回中间步骤
)

生产部署建议

性能优化方向：

• 缓存：减少重复API调用
• 批处理：Batch API提高吞吐
• 异步：Async调用降低延迟
• 流式：Streaming实时响应

架构示意：

用户请求 → API Gateway → LangChain服务 → LLM API
                              ↓
                         向量数据库 (Pinecone/FAISS)
                              ↓
                         缓存层 (Redis)

Prompt工程经验

用下来几点心得：

1. 清晰具体：别含糊，给明确指令
2. 使用分隔符：```, “”” 这些能让结构更清晰
3. 指定输出格式：要JSON就明说
4. 少样本学习：给几个例子效果很好
5. 给予思考时间：复杂任务分步骤

LangChain整体来说是个不错的框架，把LLM的调用和编排抽象得比较好。但版本迭代快，API变动多，生产用的话建议锁定版本。