一、项目核心：轻量可落地的 AI Agent 方案

本次项目聚焦LangGraph 原生能力，搭建一个支持工具调用 + 执行过程实时追踪的 AI Agent 智能助手，配套 FastAPI 后端与轻量化前端，实现端到端可交互、执行过程可观测，核心解决大模型应用落地中「工具对接不规范、执行黑盒不可控」的问题。

技术栈（核心精简版）

• 后端：Python + LangChain/LangGraph + FastAPI
• 大模型：智谱 GLM（可替换为任意兼容 LangChain 的 LLM）
• 前端：HTML + TailwindCSS + marked.js（Markdown 渲染，AI 生成后适配）
• 核心能力：标准化工具封装、LangGraph 回调追踪、前后端简单联调

二、核心实现步骤（重点代码 + 关键要点）

1. 环境准备与依赖安装

 

# 核心必装依赖
pip install langchain langgraph fastapi uvicorn python-dotenv zhipuai

新建.env文件配置大模型密钥：ZHIPU_API_KEY=你的智谱API密钥

2. 标准化工具封装（tools.py）

LangGraph 工具调用的核心规范：@tool装饰器 +完整 docstring（框架自动解析元数据），缺一不可。

from langchain_core.tools import tool
from dotenv import load_dotenv
import os
load_dotenv()

# 示例1：通用搜索工具
@tool
def std_search(date: str, query: str):
    """
    通用信息查询工具，按指定日期查询关键词相关信息
    :param date: 查询日期，格式YYYY/MM/DD（必填）
    :param query: 搜索关键词/问题（必填）
    :return: 整理后的信息+实用建议
    """
    # 实际场景可对接真实搜索接口，此处为逻辑示例
    return f"【{date}】{query}相关信息整理：（省略真实返回逻辑）"

# 示例2：邮件发送工具
@tool
def send_email(address: str, subject: str, content: str):
    """
    邮件发送工具，向指定邮箱发送邮件
    :param address: 接收邮箱（必填，如user@example.com）
    :param subject: 邮件主题（必填）
    :param content: 邮件正文（必填）
    :return: 发送结果提示
    """
    return f"已成功向{address}发送邮件，主题：{subject}"

# 可扩展：金融资讯分析、数据查询等工具，按相同规范封装

3. 大模型初始化（model.py）

极简配置，可直接替换为 OpenAI、文心一言等 LLM：

from langchain_community.chat_models import ChatZhipuAI
from dotenv import load_dotenv
import os
load_dotenv()

# 初始化智谱GLM，低温度保证回复稳定性
llm = ChatZhipuAI(
    api_key=os.getenv("ZHIPU_API_KEY"),
    model="glm-4-air",
    temperature=0.1,
    max_tokens=2000
)

4. LangGraph Agent 构建 + 执行追踪（agent.py）

核心亮点：使用 LangGraph原生 Callback 回调，实时捕获工具调用的「开始 / 完成 / 失败」，替代手动拆解 invoke，更规范、可扩展。

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langgraph.callbacks import CallbackHandler, Callbacks
from langchain_core.messages import HumanMessage
from model import llm
from tools import std_search, send_email
import uuid

# 1. 自定义回调处理器：捕获工具调用过程
class ToolTrackingCallback(CallbackHandler):
    def __init__(self):
        self.trace_logs = []
        self.current_tool = None

    def on_chain_start(self, serialized, inputs, **kwargs):
        if "messages" in inputs and isinstance(inputs["messages"][-1], HumanMessage):
            self.trace_logs.append("🤖 分析请求中，匹配最优工具...")

    def on_tool_start(self, serialized, input_str, **kwargs):
        self.current_tool = serialized.get("name", "")
        self.trace_logs.append(f"🔧 正在调用【{self.current_tool}】工具...")

    def on_tool_end(self, output, **kwargs):
        self.trace_logs.append(f"✅ 【{self.current_tool}】工具调用完成")
        self.current_tool = None

    def on_tool_error(self, error, **kwargs):
        self.trace_logs.append(f"❌ 【{self.current_tool}】调用失败：{str(error)}")
        self.current_tool = None

# 2. 创建ReAct架构Agent，启用检查点支持状态追踪
agent = create_react_agent(
    llm=llm,
    tools=[std_search, send_email],
    system_message="你是智能助手，可调用工具完成通用查询、邮件发送任务，严格按工具参数要求执行",
    checkpointer=MemorySaver()
)

# 3. 封装执行函数：返回追踪日志+最终回复
def get_agent_response(user_input):
    tracker = ToolTrackingCallback()
    config = {
        "configurable": {"thread_id": str(uuid.uuid4())},  # 唯一ID隔离请求
        "callbacks": Callbacks([tracker])
    }
    # 执行Agent
    response = agent.invoke({"messages": [HumanMessage(content=user_input)]}, config)
    final_reply = response["messages"][-1].content
    # 无工具调用时补充默认日志
    if not tracker.trace_logs:
        tracker.trace_logs = ["🤖 分析请求中...", "✅ 无需调用工具，直接生成回复"]
    return {"tool_trace": tracker.trace_logs, "final_reply": final_reply}

5. FastAPI 后端接口（server.py）

极简接口，处理请求校验 + 前后端数据交互，一键启动服务：

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
import uvicorn
from agent import get_agent_response

app = FastAPI(title="LangGraph AI Agent")
templates = Jinja2Templates(directory="templates")  # 前端模板目录

# 主页：加载前端界面
@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

# 对话接口：接收用户输入，返回追踪日志+回复
@app.post("/api/chat")
async def chat(request: Request):
    data = await request.json()
    user_msg = data.get("msg", "").strip()
    if not user_msg:
        return {"tool_trace": [], "final_reply": "⚠️ 请输入有效内容！"}
    return get_agent_response(user_msg)

# 启动服务
if __name__ == "__main__":
    uvicorn.run("server:app", host="127.0.0.1", port=2024, reload=True)

6. 前端适配（templates/index.html）

使用 AI 生成的轻量化前端代码，核心做 2 处关键适配：

1. 引入marked.js解析 AI 返回的 Markdown 格式内容，解决加粗 / 列表 / 标题不渲染问题；
2. 对接后端/api/chat接口，按时序展示tool_trace追踪日志，实现执行过程可视化（代码见文末附件，核心为 AJAX 请求 + 日志分步渲染）。

三、项目启动与核心效果

1. 启动步骤

1. 新建templates目录，放入前端index.html文件；
2. 按上述代码创建tools.py/model.py/agent.py/server.py；
3. 配置.env中的大模型密钥；
4. 运行python server.py，访问http://127.0.0.1:2024即可使用。

2. 核心效果

• 输入需求（如「2026/01/28 查询北京天气」），前端实时展示执行追踪：分析请求中→调用【std_search】工具→工具调用完成；
• AI 返回的 Markdown 格式内容（加粗 / 列表 / 标题）自动渲染，阅读体验佳；
• 工具调用失败时，前端会展示具体错误信息，便于调试。

四、关键踩坑与解决方案

1. 工具调用报错「无 docstring」：LangGraph 依赖 docstring 解析工具元数据，必须保证每个@tool装饰的函数有功能描述 + 参数说明 + 必填标识；
2. 追踪日志一闪而过：前端通过setTimeout按固定延时分步渲染tool_trace列表，模拟真实执行时序；
3. AI 返回内容格式错乱：前端引入marked.js，对 AI 返回内容做 Markdown 解析，同时适配深色 / 浅色模式样式；
4. 多用户请求冲突：为每个请求分配唯一thread_id，配合 LangGraph 的MemorySaver实现请求隔离。

五、项目扩展方向（轻量易实现）

1. 工具层：按相同规范扩展数据库查询、企业微信 / 钉钉推送、Excel 解析等工具；
2. Agent 层：添加对话记忆（langchain_core.memory），解决上下文丢失问题；
3. 性能层：改为 SSE/WebSocket 流式输出，提升实时交互体验；
4. 部署层：使用 Docker 打包项目，实现一键部署（编写简单Dockerfile即可）。

六、项目核心价值与总结

本项目是LangGraph 入门的最佳实践，核心做到 3 点：

1. 工具封装标准化：遵循 LangChain/LangGraph 规范，为后续工具扩展打下基础；
2. 执行过程可观测：利用原生 Callback 回调，实现黑盒变透明，便于调试和用户感知；
3. 端到端快速落地：FastAPI + 轻量化前端，无需复杂前端开发，快速实现可交互的产品原型。

整个项目代码精简、逻辑清晰，适合大模型应用开发入门者学习，也可作为实际业务中 AI Agent 的基础框架，按需扩展即可落地到具体场景（如智能办公、金融资讯、个人助手等）。

---

附件：前端核心适配代码（关键片段）

// 分步渲染追踪日志
function showTrace(traceList) {
    let delay = 0;
    traceList.forEach(trace => {
        setTimeout(() => {
            document.getElementById("tool-progress-text").innerText = trace;
        }, delay);
        delay += 800; // 800ms延时，模拟真实执行节奏
    });
}
// 调用后端接口
fetch("/api/chat", {
    method: "POST",
    headers: {"Content-Type": "application/json"},
    body: JSON.stringify({msg: userInput})
}).then(res => res.json()).then(data => {
    showTrace(data.tool_trace);
    document.getElementById("reply-content").innerHTML = marked.parse(data.final_reply);
});