别背八股了,手写一个Agent你就全懂了


一、背景:我被面试官问住了

事情是这样的。上个月去面一家做AI应用的中厂,面试官问了个问题:

“你了解Agent吗?如果让你从零实现一个,怎么设计?”

我当时脑子里全是"Agent = LLM + Tool + Memory"这种概念性的东西,真要说怎么落地,竟然支支吾吾说不清楚。面试官补了一刀:“你是不是只知道LangGraph调包?”

这话扎心了。确实,现在网上99%的教程都在教你怎么用LangChain/LangGraph搭Agent,但很少有人讲底层的核心架构

回来之后我痛定思痛,决定不依赖任何Agent框架,纯手写一个ReAct模式的智能体。代码写完之后,发现这东西其实没那么玄乎——核心就3个模块,今天掰开了揉碎了讲给你听。


二、最棘手的问题:Agent到底在"循环"什么?

写之前我先梳理了几个绕不开的坑:

问题1:LLM只生成文本,怎么让它"决策"?

大模型本质就是个高级文本生成器,你问它"今天天气咋样?“,它不会自己调API查天气,它只能给你写一段"建议你打开天气App看看”。咋整?

问题2:LLM怎么知道什么时候该用工具,什么时候该直接回答?

比如用户说"统计’我爱编程’有几个字",这明显需要调用字符统计函数。但如果你没告诉LLM它有这个工具,它就瞎猜。

问题3:怎么让LLM调完工具之后"记住结果",继续推理?

调一次工具拿到结果 ≠ 任务结束。比如用户问"现在北京时间几点?美国纽约几点?",LLM先调一次时间转换工具拿到结果,然后还得基于这个结果做二次推理才能回答用户。多轮推理的上下文怎么维护?

这三个问题,本质上就是ReAct模式要解决的核心问题


三、深度分析:ReAct模式到底是个啥?

ReAct = Reasoning + Acting,翻译成人话就是思考→行动→观察→再思考的循环。

打个比方:你做饭的时候——

  • Thought(思考): “菜快糊了,我得关小火”
  • Action(行动): 伸手去拧灶台旋钮
  • Observation(观察): 火变小了,菜不糊了
  • Final Answer(最终结论): 菜炒好了,出锅

Agent也是一样的道理,只不过它的"行动"是调用工具函数,它的"观察"是函数返回的结果。

整个循环长这样:

循环开始
  ↓
① 组装Prompt(问题 + 历史记录 + 工具列表)
  ↓
② 扔给LLM,让它输出"思考"或者"最终答案"
  ↓
③ 如果是"最终答案" → 结束,返回结果
  ↓
④ 如果是"行动指令" → 解析出工具名 + 参数
  ↓
⑤ 执行工具 → 拿到结果(Observation)
  ↓
⑥ 把本轮(思考, 行动, 观察)存入历史 → 回到①

明白了这个流程,后面撸代码就是顺着这个逻辑一步一步写。


四、一步步排查踩坑:我当初是怎么写崩的

踩坑1:模型输出格式不稳定

一开始我把Prompt写得比较随意,结果LLM返回的东西五花八门——

  • 有时候返回 Action: count_characters 但少了 Action Input:
  • 有时候"Final Answer"和"Action"都出现在同一段输出里
  • 有时候回答了但不带"Final Answer:"前缀

解决方案: 在System Prompt里要求LLM严格二选一输出——要么出完整的"Thought + Action + Action Input",要么出"Final Answer",不能两种混着来。然后代码里用 parse_action() 做防御性解析,没解析到就报错提示。

踩坑2:LLM"记不住"前几轮干了啥

每次循环只把"当前问题"扔给LLM,结果第二轮LLM就忘了刚才工具返回的结果是什么。相当于你跟人说话,每次说一半就重新开始——

解决方案: 维护一个 history: List[Tuple[thought, action_info, observation]],每轮执行完都追加进去,下次组装Prompt时一并带上。这样LLM每次都是"带着前几轮的完整记录"做决策。

踩坑3:工具注册过于耦合

最开始我把工具调用写死在业务逻辑里,加一个新工具就得改loop代码,很蠢。

解决方案: 定义统一的 Tool 数据结构(名字 + 描述 + 执行函数),外部统一注册成字典传入。loop完全不用关心工具内部实现,只负责"名字匹配→执行→拿结果"。


五、最终实现方案 + 完整代码

好,废话不多说,上代码。整个项目分为3个核心模块,结构如下:

agent_test/
├── main.py                  # 入口:组装LLM、注册工具、启动循环
├── agent/
│   ├── types.py             # 统一工具数据结构
│   ├── prompt.py            # Prompt构建 + 输出解析
│   ├── loop.py              # ReAct核心循环
│   └── tools/
│       ├── word_count.py    # 工具1:字符统计
│       └── covert.py        # 工具2:时间转换
└── llm/
    └── deepSeek.py          # LLM封装

模块1:统一工具数据结构(types.py)

所有工具的"标准接口",不管什么工具都得遵守这个格式:

from dataclasses import dataclass
from typing import Callable

@dataclass
class Tool:
    name: str                                     # 工具名
    description: str                              # 工具描述(给LLM看)
    run: Callable[[str], str]                     # 执行函数

为什么要有 description?因为LLM是通过文字描述来理解工具的。工具名叫 count_characters,描述写"计算字符串字符数量",LLM看到用户说"统计几个字",就知道调用它。

模块2:Prompt构建与解析(prompt.py)

这个模块干两件事:

第一件事:把系统指令、工具信息、历史记录拼成一段发给LLM

def build_prompt(question, history, tools):
    # 1. 把所有工具的name和description拼成文本
    tool_text = ""
    for name, tool in tools.items():
        tool_text += f"{name}:{tool.description}\n"

    # 2. 把前几轮的推理记录拼成文本
    history_text = ""
    for thought, action_info, obs in history:
        history_text += f"""
Thought: {thought}
Action: {action_info}
Observation: {obs}
"""

    # 3. 基座Prompt + 工具清单 + 历史记录 + 当前问题
    full_prompt = f"{base_prompt}{tool_text}{history_text}\n用户问题:{question}"
    return full_prompt

第二件事:从LLM的回复里解析出"工具名"和"参数"

def parse_action(text):
    action = ""
    input_val = ""
    for line in text.splitlines():
        if line.startswith("Action:"):
            action = line.split("Action:", 1)[1].strip()
        if line.startswith("Action Input:"):
            input_val = line.split("Action Input:", 1)[1].strip()
    return action, input_val

这里基本逻辑就是逐行扫描,看到 Action: 开头的行就提取工具名,看到 Action Input: 开头的行就提取入参。简单粗暴但够用。

模块3:ReAct核心循环(loop.py)

这是整个Agent的心脏,前面说的思考→行动→观察重复循环就在这里:

def react_loop(question, tools, llm, max_steps=10, verbose=True):
    history = []                          # 存放多轮交互记录

    for i in range(1, max_steps + 1):
        # ① 组装完整Prompt
        prompt = build_prompt(question, history, tools)

        # ② 调用LLM
        llm_output = llm(prompt)

        # ③ 如果LLM给出最终答案 → 直接返回
        if "Final Answer:" in llm_output:
            return llm_output.split("Final Answer:", 1)[1].strip()

        # ④ 否则解析出工具名和参数
        tool_name, tool_input = parse_action(llm_output)

        # ⑤ 校验工具是否存在
        if not tool_name or tool_name not in tools:
            observation = f"ERROR: 你选的工具不存在,可用工具只有 {list(tools.keys())}"
        else:
            observation = tools[tool_name].run(tool_input or "")

        # ⑥ 把本轮记录追加到历史
        history.append((thought, f"{tool_name}[{tool_input}]", observation))

    return "Failed: 达到最大循环次数,无法解答该问题"

核心要点:

  • history是List[Tuple],每轮追加(思考, 行动, 观察),下一轮LLM能看到前因后果
  • max_steps兜底,防止LLM无限循环调用工具
  • 工具校验,防止LLM幻觉调用不存在的工具

具体工具实现

字符统计工具:

from agent.types import Tool

def count_characters(text: str) -> str:
    text = text.strip()
    length = len(text)
    return str(length)

word_count_tool = Tool(
    name="count_characters",
    description="计算字符串的字符数量,返回数字",
    run=count_characters
)

时间转换工具:

from datetime import datetime
from zoneinfo import ZoneInfo
from agent.types import Tool

def covert_time(time: str) -> str:
    cn_now = datetime.now(ZoneInfo("Asia/Shanghai"))
    us_now = cn_now.astimezone(ZoneInfo("America/New_York"))
    return (f"北京时间:{cn_now.strftime('%Y-%m-%d %H:%M:%S')}\n"
            f"美国纽约时间:{us_now.strftime('%Y-%m-%d %H:%M:%S')}")

time_tool = Tool(
    name="covert_time",
    description="将北京时间转化成美国纽约时间",
    run=covert_time
)

LLM封装(deepSeek.py)

把DeepSeek API包装成 Callable[[str], str],输入Prompt返回文本:

from openai import OpenAI

def create_deepseek_llm(model="deepseek-chat", api_key=None, base_url="https://api.deepseek.com"):
    key = api_key or os.environ.get("DEEPSEEK_API_KEY")
    client = OpenAI(api_key=key, base_url=base_url)

    def llm(prompt: str) -> str:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0,            # 温度设为0,输出一致性高
        )
        return response.choices[0].message.content or ""

    return llm

注意 temperature=0——Agent场景下我们不需要LLM"创意发挥",需要的是稳定、可预测的输出格式,温度越低越好。

入口文件(main.py)

把所有模块串起来:

from llm.deepSeek import create_deepseek_llm
from agent.loop import react_loop
from agent.tools.word_count import word_count_tool
from agent.tools.covert import time_tool

# 创建LLM(从环境变量读API Key)
llm_func = create_deepseek_llm()

# 注册工具
tools = {
    "count_characters": word_count_tool,
    "covert_time": time_tool
}

if __name__ == "__main__":
    question = "统计文本:我爱编程一共有多少字符"
    res = react_loop(question, tools, llm_func, verbose=True)
    print("最终回答:", res)

六、运行效果:看看它怎么"思考"的

开启 verbose=True 后,你能看到每一轮LLM的完整思考过程:

第一轮:

LLM原始输出:
Thought: 用户想统计"我爱编程"的字符数,我需要使用count_characters工具
Action: count_characters
Action Input: 我爱编程

🔧 工具执行完毕,返回结果Observation:4

第二轮(带历史重新问):

LLM原始输出:
Thought: 工具返回了4,说明"我爱编程"有4个字符,直接回答用户
Final Answer: "我爱编程"一共有4个字符

最终回答: ✅ AI信息足够,直接给出最终答案:"我爱编程"一共有4个字符

看到了吗?LLM自己会"想"——第一轮发现需要工具,就调工具;拿到结果后第二轮发现信息够了,就直接回答。整个过程不需要你写if-else判断。


七、总结 + 避坑技巧

核心3句话总结

  1. Agent = ReAct循环:思考→行动→观察→再思考,直到能给出答案
  2. LLM通过Prompt中的文字描述理解工具:工具的描述写得好不好,直接决定LLM会不会正确调用
  3. 历史记录是关键:没有历史,LLM每轮都是"失忆"状态,无法做多步推理

避坑清单(面试/实战都用得上)

后果 怎么避
Tool描述太随意 LLM不知道该调用哪个工具 描述写清楚"什么时候用、干什么用"
忘记传历史记录 LLM每轮都重头开始 每次Prompt必须拼接history
没设max_steps LLM可能无限循环(烧钱!) 默认10轮上限
temperature太高 输出格式不稳定,解析失败 Agent场景设0或接近0
工具名硬编码 加工具就得改代码 统一Tool数据结构,字典注册
不校验工具名 LLM幻觉调了不存在的工具崩了 先check tool_name in tools

给面试的你

如果面试官问你"你怎么实现Agent",直接按我这套说:

  1. 定义统一的Tool接口(name, description, run)
  2. 设计System Prompt,要求LLM按固定格式输出Action/Thought/Final Answer
  3. 实现ReAct循环,维护history,LLM输出→解析→执行→存历史→再问LLM
  4. 循环直到出现Final Answer或超max_steps

这比背LangChain源码强一百倍——因为这证明你真的理解Agent的本质而不只是会调包


写在后面: 这个项目是我在 D:\pyTest\agent_test 写的demo,所有代码加起来不到200行,但Agent的"魂"全在里面了。想跑起来的同学,装个 openai 库,配个DeepSeek的API Key就能玩。

下一期我计划在这个基础上加上RAG(检索增强生成),让Agent能"读取本地文档"再回答问题,感兴趣的点个关注,更着不迷路。


如果你觉得这篇文章对你有帮助,点赞收藏转发走一波,让更多被LangChain折磨的同学看到真正的Agent长啥样。有问题评论区见,看到就回。

更多推荐