大模型时代：如何用Python构建智能Agent系统

1. 为什么现在人人都想搓个Agent？

最近半年，我身边搞技术的朋友见面第一句话已经从"吃了吗"变成"你的Agent跑通了吗"。这种狂热让我想起2017年区块链火爆时，连楼下卖煎饼的大爷都在问"要不要看看我的白皮书"。但这次不一样——大模型确实让构建智能体（Agent）的门槛降到了历史最低点。

上周帮一个产品经理调试他"手搓"的会议纪要生成Agent时，我震惊地发现：用不到200行Python代码就能实现自动记录、要点提炼和待办事项跟踪。这要放在三年前，至少需要一个NLP团队折腾三个月。现在，我们正处在一个奇妙的拐点：理解Agent的工作原理比实现它更难。

2. 解剖麻雀：一个最小可行Agent的四大器官

2.1 大脑皮层：LLM核心

我用OpenAI的gpt-3.5-turbo作为基础模型时，发现三个关键参数决定了Agent的"智商"：

• temperature=0.3（保持适度创造性）
• max_tokens=512（防止话痨）
• stop_sequence=["\nObservation:"]（关键！控制交互边界）

python复制from openai import OpenAI
client = OpenAI()

def think(prompt, memory):
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "system", "content": memory},
                 {"role": "user", "content": prompt}],
        temperature=0.3,
        max_tokens=512,
        stop=["\nObservation:"]
    )
    return response.choices[0].message.content

2.2 海马体：记忆系统

短期记忆用列表缓存最近5轮对话，长期记忆我推荐用Chroma这类轻量级向量数据库。实测下来，记忆检索的top_k设为3时性价比最高：

python复制import chromadb
chroma_client = chromadb.Client()
memory_collection = chroma_client.create_collection(name="memory")

def remember(texts):
    ids = [str(uuid.uuid4()) for _ in texts]
    memory_collection.add(documents=texts, ids=ids)

def recall(query):
    return memory_collection.query(query_texts=[query], n_results=3)

2.3 小脑：工具调用

给Agent安装"手"和"眼睛"的关键在于工具描述。我发现用三句式模板效果最好：

1. 工具名：<calculate>
2. 功能：Performs math calculations
3. 输入示例：<calculate>(3+5)*2

python复制tools = {
    "calculate": lambda x: str(eval(x)),
    "search": google_search  # 需预先封装
}

def use_tool(tool_name, params):
    return tools.get(tool_name, lambda _: "Tool not found")(params)

2.4 脑干：循环控制

最容易被忽视的是决策循环。我设计的状态机包含三个关键状态：

1. THINKING：分析用户输入
2. ACTING：执行工具调用
3. SLEEPING：等待唤醒信号

python复制class AgentState:
    def __init__(self):
        self.state = "SLEEPING"
        self.last_action = None

    def transition(self, new_state):
        valid_transitions = {
            "SLEEPING": ["THINKING"],
            "THINKING": ["ACTING", "SLEEPING"],
            "ACTING": ["THINKING"]
        }
        if new_state in valid_transitions[self.state]:
            self.state = new_state
            return True
        return False

3. 从玩具到工具：避开三个致命陷阱

3.1 幻觉抑制：给模型戴上"缰绳"

在测试电商客服Agent时，遇到过模型擅自承诺"24小时到货"的严重事故。现在我会强制在系统提示中加入：

text复制你必须严格遵守以下规则：
1. 不知道就说"需要查证"
2. 涉及时间/金额必须确认
3. 不能自行补充事实细节

配合正则表达式校验输出：

python复制import re

def safety_check(text):
    danger_patterns = [
        r"\b\d+小时?到货\b",
        r"\b百分之百\b",
        r"\b绝对\b"
    ]
    return not any(re.search(p, text) for p in danger_patterns)

3.2 上下文管理：记忆的黄金分割

当对话轮次超过7轮时，我发现模型开始出现"记忆混淆"。现在的解决方案是：

1. 每5轮做一次摘要
2. 将摘要存入长期记忆
3. 清空短期记忆缓冲区

python复制def summarize(messages):
    summary_prompt = "用100字总结以下对话要点：\n" + "\n".join(messages)
    return think(summary_prompt, "你是一个专业的摘要生成器")

3.3 成本控制：别让API调用破产

早期版本我的天气查询Agent因为忘记加限流，一晚上烧了$80。现在必做三件事：

1. 令牌计数：len(text.split()) * 1.33（安全系数）
2. 请求限速：@rate_limit(3, 60) # 每分钟3次
3. 熔断机制：错误率>5%时自动休眠

python复制from functools import wraps
import time

def rate_limit(calls, period):
    def decorator(func):
        history = []
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            history[:] = [t for t in history if t > now - period]
            if len(history) >= calls:
                raise RuntimeError("Rate limit exceeded")
            history.append(now)
            return func(*args, **kwargs)
        return wrapper
    return decorator

4. 实战：构建会议秘书Agent

4.1 需求拆解

帮市场部做的Agent需要实现：

• 实时转录（用Whisper API）
• 提取决议项（关键动作识别）
• 生成待办（SMART原则格式化）

python复制def parse_meeting(audio):
    transcript = transcribe(audio)  # 调用语音转文本
    decisions = think(
        "提取以下会议记录中的关键决议：\n" + transcript,
        "你擅长识别会议中的决策点"
    )
    todos = []
    for line in decisions.split("\n"):
        if "]" in line:  # 检测到任务项
            todos.append(line.split("]")[1].strip())
    return {"transcript": transcript, "todos": todos}

4.2 效果优化技巧

经过20次真实会议测试，总结出三个提升点：

1. 人名消歧：提前注入参会者名单
2. 时间解析：统一转换"下周三"为具体日期
3. 责任归属：用正则匹配"@张三负责"

python复制participants = ["张三", "李四", "王五"]

def clarify_names(text):
    for name in participants:
        text = text.replace(f"{name[0]}总", name)
    return text

def parse_time(text):
    from dateparser import parse
    return parse(text, settings={'PREFER_DATES_FROM': 'future'})

4.3 部署踩坑记录

在Docker化部署时遇到的典型问题：

1. 异步调用卡死：必须用async/await重构
2. 长耗时任务：改用Celery任务队列
3. 内存泄漏：发现是ChromaDB的缓存问题，改为每24小时重启

dockerfile复制FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["gunicorn", "-k", "uvicorn.workers.UvicornWorker", "main:app"]

5. 进阶路线：从单兵作战到多Agent协同

当基本Agent跑通后，可以尝试：

1. 角色扮演：让多个Agent模拟市场/技术/财务部门辩论
2. 联邦学习：用LoRA技术让Agents互相学习
3. 自进化：定期让Agent分析自己的错误日志

python复制def role_play(agents, scenario):
    for i in range(3):  # 三轮辩论
        for agent in agents:
            response = agent.think(scenario)
            scenario += f"\n{agent.role}观点:{response}"
    return think("总结各方论点", scenario)

最近在实验的"Agent孵化器"模式很有趣：主Agent负责诊断子Agent的问题并自动修改代码。有次它甚至给自己加了缓存机制——虽然加的位置不对导致更多bug，但这个方向值得探索。