LangChain 实战指南：把关键能力落到项目里

头像违规？？

156人浏览 · 2026-08-06 20:52:58

头像违规？？ · 2026-08-06 20:52:58 发布

聊《LangChain怎么学？先做一个会暴露问题的真实项目》之前，先说一句实在的：别急着背概念，先看它在真实项目里到底解决什么问题。

摘要

LangChain生态很火，但很多开发者学完后发现，自己的Agent项目只能跑个Demo，真正上线就各种问题。本文结合真实项目经验，从核心组件到工具调用，再到生产环境的关键考量，分享LangChain实战中的坑和解决方案。

LangChain能解决什么问题

文章插图 1

先说个真实场景。去年有个做运维的同学找我，说想做个Agent自动处理告警。Demo跑得很顺，模型能读监控数据、能查历史日志、还能生成修复建议。但真正要接入公司系统时，问题全来了——API调用没权限、日志没记录、模型响应时间不可控。

LangChain解决的问题，本质上是把散乱的AI能力串成可复用的工程。它不是让你"学会用大模型"，而是帮你解决：

1. Prompt管理：不同场景的提示词怎么版本化、怎么测试
2. Chain编排：多个模型调用之间的数据怎么流转
3. 工具集成：怎么让模型调用外部API、数据库、文件系统
4. 记忆管理：多轮对话的状态怎么保存和恢复

很多人学LangChain，第一步就错了——先学怎么调API，再学怎么拼Chain。其实应该反过来，先想清楚你的Agent要做什么，再决定用什么组件。

核心组件

文章插图 2

LangChain的组件很多，但真正常用的就这几类：

模型层

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.1,
    max_tokens=2048,
    timeout=30,
    max_retries=2
)

temperature别设太高，生产环境要的是稳定输出，不是创意。

提示词层

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个运维助手，负责处理告警和生成修复方案。"),
    ("user", "当前告警信息：{alert}\n历史日志：{logs}\n请给出处理建议。")
])

提示词要模块化，别把所有逻辑塞进一个字符串。

链式调用

from langchain_core.output_parsers import JsonOutputParser

parser = JsonOutputParser()
chain = prompt | llm | parser

这里用了管道语法，比LLMChain更清晰。

记忆管理

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

class SimpleMemory:
    def __init__(self, max_turns=10):
        self.history = []
        self.max_turns = max_turns

    def add(self, role, content):
        self.history.append({"role": role, "content": content})
        if len(self.history) > self.max_turns * 2:
            self.history = self.history[-self.max_turns * 2:]

    def get_messages(self):
        return [
            HumanMessage(content=m["content"]) if m["role"] == "user"
            else AIMessage(content=m["content"])
            for m in self.history
        ]

别迷信LangChain内置的记忆组件，简单场景自己写更可控。

CSDN资料领取方式

Prompt与Chain

Prompt工程是LangChain的核心，但也是最容易踩坑的地方。

常见错误

提示词太长，模型注意力分散
没有明确输出格式，解析失败
忘记考虑token限制，实际调用时截断

我的做法

from langchain_core.prompts import PromptTemplate

# 分离系统提示和用户提示
SYSTEM_PROMPT = """
你是一个智能客服助手。你的职责：
1. 理解用户问题
2. 从知识库中查找相关信息
3. 给出准确、简洁的回答

约束：
- 如果不知道答案，直接说"我不知道"
- 回答控制在200字以内
- 不要编造信息
"""

USER_TEMPLATE = PromptTemplate(
    input_variables=["question", "context"],
    template="""
知识库内容：
{context}

用户问题：{question}

请根据知识库内容回答问题。
"""
)

Chain的调试技巧

from langchain_core.runnables import RunnablePassthrough

debug_chain = {
    "question": RunnablePassthrough(),
    "context": lambda x: search_knowledge(x["question"])
} | USER_TEMPLATE | llm

# 调试时打印中间结果
result = debug_chain.invoke({"question": "如何重置密码？"})
print(result)

用RunnablePassthrough和字典语法，可以方便地调试每个环节的输出。

工具调用

工具调用是Agent的"手脚"，也是生产环境最容易出问题的地方。

定义工具

from langchain_core.tools import tool
import requests

@tool
def get_system_status(service_name: str) -> dict:
    """查询指定服务的系统状态"""
    response = requests.get(
        f"https://monitor.example.com/api/status/{service_name}",
        headers={"Authorization": "Bearer YOUR_TOKEN"},
        timeout=10
    )
    return response.json()

@tool
def execute_command(command: str, target_host: str) -> str:
    """在指定主机上执行命令（需要审批）"""
    # 这里应该加权限校验
    if not check_permission(target_host):
        return "权限不足"
    result = run_remote_command(target_host, command)
    return result

工具调用的坑
1. 权限问题：别把敏感操作直接暴露给模型，要有审批机制
2. 超时处理：网络请求可能卡住，要设timeout
3. 错误恢复：工具调用失败时，模型可能陷入循环，要加最大重试次数

调用工具的正确姿势

from langchain.agents import create_openai_functions_agent, AgentExecutor

tools = [get_system_status, execute_command]
agent = create_openai_functions_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,  # 防止无限循环
    handle_parsing_errors=True,
    verbose=True
)

result = executor.invoke({"input": "检查nginx状态并重启"})

项目实战

去年我做了一个运维Agent项目，从Demo到生产，踩了不少坑。

第一阶段：Demo跑通

from langchain.agents import initialize_agent, AgentType
from langchain.tools import Tool

def check_disk_usage(host: str) -> str:
    # 模拟API调用
    return f"{host} 磁盘使用率: 75%"

tools = [
    Tool(
        name="check_disk",
        func=check_disk_usage,
        description="查询主机磁盘使用情况"
    )
]

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.OPENAI_FUNCTIONS,
    verbose=True
)

agent.run("检查web01的磁盘使用情况")

Demo阶段，所有东西都是硬编码，跑通了就以为结束了。

第二阶段：接入真实系统
问题开始暴露：

API调用超时，Agent卡死
模型输出格式不固定，解析失败
没有日志记录，出问题无法排查

解决方案

import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("agent")

class TracedAgent:
    def __init__(self, executor):
        self.executor = executor

    def invoke(self, input_text: str) -> dict:
        trace = {
            "timestamp": datetime.now().isoformat(),
            "input": input_text,
            "steps": []
        }

        try:
            result = self.executor.invoke({"input": input_text})
            trace["output"] = result
            trace["status"] = "success"
        except Exception as e:
            trace["error"] = str(e)
            trace["status"] = "failed"
            logger.error(f"Agent execution failed: {e}")

        trace["steps"] = self.executor.history
        logger.info(f"Agent trace: {trace}")

        return trace

加上日志和trace后，问题排查效率提升了至少50%。

第三阶段：生产环境
最后的问题是关于权限和可观测性：

敏感操作需要人工审批
调用量监控和成本核算
模型响应时间的SLA保障

import time
from functools import wraps

def track_performance(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            duration = time.time() - start
            logger.info(f"{func.__name__} took {duration:.2f}s")
            return result
        except Exception as e:
            duration = time.time() - start
            logger.error(f"{func.__name__} failed after {duration:.2f}s: {e}")
            raise
    return wrapper

这些看似简单的改进，才是Demo和生产环境的真正差距。