AI Agent与大模型开发入门：从零构建你的第一个智能代理

1次阅读

没有评论

共计 3313 个字符，预计需要花费 9 分钟才能阅读完成。

AI Agent（智能代理）可以理解为一个能自主决策的软件程序，它通过感知环境、分析信息、执行动作来实现特定目标。大模型（如 GPT-4）则是它的 ” 大脑 ”，负责理解语言、生成响应和逻辑推理。

举个生活中的例子：AI Agent 就像一名餐厅服务员，大模型是其掌握的菜品知识和沟通能力。服务员（Agent）需要根据顾客需求（输入），结合自己的知识（大模型），决定如何回答问题或采取行动（输出）。

过度依赖大模型：试图让大模型完成所有工作，忽视业务逻辑的独立封装
忽视状态管理：未设计对话历史记录机制，导致多轮对话上下文丢失
盲目追求模型规模：误认为模型参数量越大越好，忽略推理成本和延迟问题
缺乏错误处理：未考虑 API 调用失败、网络波动等异常场景

安装 Python 3.8+ 并创建虚拟环境

安装依赖库：

pip install langchain openai python-dotenv

在项目根目录创建 .env 文件存储 API 密钥：
```
OPENAI_API_KEY=your_key_here
```

from langchain.agents import initialize_agent, AgentType
from langchain.llms import OpenAI
from langchain.chat_models import ChatOpenAI
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

# 初始化大模型（建议使用 gpt-3.5-turbo 平衡成本与效果）llm = ChatOpenAI(
    temperature=0.5,  # 控制输出随机性
    model_name="gpt-3.5-turbo"
)

# 创建基础 Agent
agent = initialize_agent(
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    llm=llm,
    verbose=True,  # 打印详细执行过程
    handle_parsing_errors=True  # 自动处理解析错误
)

# 执行简单查询
response = agent.run("用中文解释什么是机器学习？")
print(response)

关键参数说明：
– AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION：适合多轮对话的 Agent 类型
– temperature：值越低输出越确定，适合事实性问答；值越高创造性越强

实现对话历史记忆的两种方式：

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(memory_key="chat_history")
agent = initialize_agent(
    # ... 其他参数同前...
    memory=memory
)

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.docstore import InMemoryDocstore

# 创建向量数据库
embeddings_model = OpenAIEmbeddings()
vectorstore = FAISS(embeddings_model, InMemoryDocstore({}), {})

# 创建带记忆的 Agent
from langchain.agents import AgentExecutor
from langchain.agents.react.base import ReActDocstoreAgent

tools = [...]  # 定义你的工具集
agent = ReActDocstoreAgent.from_llm_and_tools(llm, tools)
agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent,
    tools=tools,
    memory=vectorstore.as_retriever())

使用 max_tokens 参数限制响应长度
对大文本响应启用 streaming 流式输出
对历史对话进行摘要而不是完整存储

示例代码：

llm = ChatOpenAI(
    max_tokens=500,
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

import asyncio
from langchain.agents import load_tools

async def async_agent_query(query):
    tools = load_tools(["serpapi"], llm=llm)
    agent = initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION)
    return await agent.arun(query)

# 并发执行多个查询
tasks = [async_agent_query(q) for q in questions]
results = await asyncio.gather(*tasks)

实现指数退避重试机制：

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
)

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def safe_ai_call(prompt):
    return agent.run(prompt)

监控 Token 使用量：

from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = agent.run("你的查询")
    print(f"本次消耗 Token: {cb.total_tokens}")

使用正则表达式过滤输出：

import re

def sanitize_output(text):
    return re.sub(r'\b(?:password|api[_-]?key)\b', '[REDACTED]', text, flags=re.IGNORECASE)

集成专业过滤库：

pip install presidio-analyzer presidio-anonymizer

实现检查点保存：

import pickle

def save_checkpoint(agent, filepath):
    with open(filepath, 'wb') as f:
        pickle.dump({
            'memory': agent.memory,
            'tools': agent.tools
        }, f)

超时自动取消：

from func_timeout import func_timeout, FunctionTimedOut

try:
    result = func_timeout(30, agent.run, args=("查询",))
except FunctionTimedOut:
    print("请求超时，已终止")