🔥【Agent】应用全流程

Prompt

经过十几个月的发展，提示词工程中技巧性的部分已经不重要了。现在的模型唯一刚需是把非通用的背景知识、目的等写清楚。

RTF 框架可以继续使用，这仍然是一个还不错的格式化书写方法。

调用大模型

推荐 HTTP 方式调用（使用 request）

密钥管理：

# Linux：
echo 'export LLM_API_KEY="你的真实API_KEY"' >> ~/.bashrc
source ~/.bashrc


# MacBook：
echo 'export LLM_API_KEY="你的真实API_KEY"' >> ~/.zshrc
source ~/.zshrc

# 测试已经成功设置
test -n "$LLM_API_KEY" && echo "API Key 已设置"

import os
import requests

api_key = os.environ["LLM_API_KEY"]
endpoint_url = "https://xxx.xxx.com/v1/chat/completions" # 注意结尾是 /v1/chat/completions


class LLMClient:
    def __init__(self, api_key, endpoint_url):
        self.endpoint_url = endpoint_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

    def call(self, payload):
        response = requests.post(
            self.endpoint_url,
            headers=self.headers,
            json=payload,
            timeout=(5, 120),
        )
        response.raise_for_status()
        return response.json()

llm_client = LLMClient(api_key, endpoint_url)

普通调用

model_name = 'DeepSeek-V4-Flash-0731'

payload = {
    'model': model_name,
    'messages': [
        {'role': 'system', 'content': '你的任务是一句话解释一个概念'},
        {'role': 'user', 'content': '解释 yolo'}
    ]
}

resp = llm_client.call(payload)

print(resp)
print(resp['choices'][0]['message'])

print(resp['choices'][0]['message']['content'])

# 其它可能有用的信息：
# resp.choices[0]['message']['role']
# resp.choices[0]['message']['reasoning_content']
# resp.choices[0]['message']['function_call']
# resp.choices[0]['message']['tool_calls']

多模态

model_name = 'Qwen3-VL-235B-A22B-Instruct'

import base64


def b64_image(path):
    with open(path, "rb") as f:
        return base64.b64encode(f.read()).decode("ascii")


payload = {
    'model': model_name,
    'messages': [
        {'role': 'system', 'content': "你是一个看图专家"},
        {'role': 'user', 'content': [
            {"type": "text", "text": "这张图里有什么？"},
            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64_image('1.png')}"}},
            # {"type": "image_url", "image_url": {"url": "https://xxx/1.png"}},
        ]}
    ]

}

resp = call_llm.call(payload)

print(resp)

embedding

# 注意，这里的 endpoint 不同
llm_client = LLMClient(api_key, endpoint_url="https://xxx.xxx.com/v1/embeddings")

model_name = 'Qwen3-Embedding-8B'

payload = {
    'model': model_name,
    "input": [
        "外卖不好吃"
        , "送的很快"
        , "太油了，太辣了"],

}

resp = llm_client.call(payload)

item = resp['data'][0]
item['index'], item['embedding']  # list<float>

流式输出

import json
import requests

payload = {
    "model": "DeepSeek-V3.2",
    "messages": [{"role": "user", "content": "写一个Python函数，判断素数。"}],
    "stream": True,
}

with requests.post(
    base_url,
    headers={"Authorization": f"Bearer {api_key}"},
    json=payload,
    stream=True,
    timeout=(5, 120),
) as response:
    if not response.ok:
        raise RuntimeError(f"HTTP {response.status_code}: {response.text}")

    for raw_line in response.iter_lines(chunk_size=1):
        if not raw_line:
            continue

        line = raw_line.decode("utf-8")
        if not line.startswith("data:"):
            continue

        data = line[5:].strip()
        if data == "[DONE]":
            break

        chunk = json.loads(data)
        choices = chunk.get("choices") or []
        if choices:
            content = (choices[0].get("delta") or {}).get("content")
            if content:
                print(content, end="", flush=True)

print()

一些有用的参数

# 概率相关：
"top_p": 0.2,
"temperature": 0.1,
"frequency_penalty": 0,
"presence_penalty": 0,

# 最大输出 token：
"max_tokens": 4096,

# 流式输出
"stream": false,

# 开启/关闭思考
"chat_template_kwargs": {"enable_thinking": False},

输出限制 JSON

# 输出格式限制为 json：
"response_format": {
    "type": "json_object"
}
# 如果想要更多的 json 格式限制，可以这样：
payload["response_format"] = {
    "type": "json_schema",
    "json_schema": {
        "name": "concept_explanation",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "概念名": {"type": "string", "description": "需要解释的概念名称"},
                "解释": {"type": "string", "description": "简明解释，不超过15字"},
                "归类": {"type": "string", "description": "该概念所属的知识领域"},
                "难度": {"type": "string", "enum": ["入门", "基础", "中等", "进阶"], "description": "学习该概念的难度"}
            },
            "required": ["概念名", "解释", "归类", "难度"],
            "additionalProperties": False
        }
    }
}

# 注意
# 1. description 也会起作用（大部分模型），与提示词的作用差不多
# 2. 不一定对所有模型都兼容，使用前注意测试

tools 的例子

import json



def get_weather(city):
    # 演示数据，不代表实时天气
    weather = {
        "北京": {"condition": "小雨", "temperature_c": 18},
        "上海": {"condition": "晴", "temperature_c": 26},
    }
    return weather.get(city, {"error": "暂无该城市的天气数据"})


tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询指定城市的天气和气温",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称"}
            },
            "required": ["city"],
        },
    },
}]

messages = [
    {"role": "user", "content": "查一下北京的天气，并给我一条穿衣建议。"}
]

for _ in range(3):
    resp = llm_client.call({
        "model": "DeepSeek-V3.2",
        "messages": messages,
        "tools": tools,
        "tool_choice": "auto",
    })

    message = resp["choices"][0]["message"]
    tool_calls = message.get("tool_calls") or []

    if not tool_calls:
        print(message["content"])
        break

    messages.append(message)

    for tool_call in tool_calls:
        function = tool_call["function"]
        if function["name"] != "get_weather":
            raise ValueError(f"未知工具：{function['name']}")

        arguments = json.loads(function["arguments"])
        result = get_weather(arguments["city"])

        messages.append({
            "role": "tool",
            "tool_call_id": tool_call["id"],
            "content": json.dumps(result, ensure_ascii=False),
        })
else:
    raise RuntimeError("工具调用次数过多")

Agent

Agent 是包含了 RAG、提示词、Function calling、Re-Act等一系列元素的集合

API 调用，例如查询天气、发送邮件
Re-Act

自主Agent

由 Planner 模块（也是个LLM），与其它 LLM、工具调用的配合

Agent 优化

GEPA (Genetic-Pareto)
- 边界内收敛到最优
- 优化 prompt
Auto Research
- 开放的边界，探索高价值改动方向
- 优化整个工作流
Agentic RL (Agentic Reinforcement Learning)
- 根据完整轨迹，对模型做后训练

context 构建

Context 包括什么

System Prompt
- 定义角色、能力、行为约束
User Goal
- 用户的原始目标
Task State
- 目前的执行状态
Conversation History
- 历史对话，它是无限增长的，因此需要压缩（Summarization）
Memory：长期应该知道什么
- Profile Memory，用户画像、偏好等
- Episodic Memory，历史事件。之前解决的问题、踩过的坑
- Semantic Memory，最佳实践、python 包
Retrieved Knowledge
- RAG 得到的信息，例如从1000个文件中检索到的 2个相关文件
Tool Description
- 有哪些工具可供调用
Tool Results
- 工具返回结果
Previous Reasoning
- 上一步的推理状态

一个最小完整的 agent 例子

这一步这与前面一模一样

api_key = "xxx" # 同上的密钥管理
base_url="https://xxx.xxx.com/v1/chat/completions" # 注意结尾是 /v1/chat/completions

import requests

class LLMClient:
    def __init__(self, api_key, endpoint_url):
        self.endpoint_url = endpoint_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

    def call(self, payload):
        response = requests.post(
            self.endpoint_url,
            headers=self.headers,
            json=payload,
            timeout=(5, 120),
        )
        response.raise_for_status()
        return response.json()

call_llm = LLMClient(api_key, base_url)

定义大模型可以调用的工具：

tools = [
    # shell 工具，LLM 根据通用知识确定如何写 shell 命令，而不是提供大量的、地球人都知道的 Linux 命令
    {
        "type": "function",
        "function": {
            "name": "terminal",
            "description": """
            Execute a shell command in the local environment.
            The model should generate appropriate shell commands
            based on its existing knowledge of Linux CLI tools.
            """,
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": "Shell command to execute"
                    }
                },
                "required": [
                    "command"
                ]
            }
        }
    }

    # 自定义 python 函数，明确告诉 LLM 这个函数如何使用
    , {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的天气和气温",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"],
            },
        },
    }
]

import subprocess
import json


def terminal(command):
    """执行 shell 命令，
    没有做安全过滤，请不要让 agent 执行危险动作"""
    result = subprocess.run(
        command,
        shell=True,
        capture_output=True,
        text=True
    )

    return result.stdout + result.stderr


def get_weather(city):
    """模拟天气 API"""
    return f"{city} 今天晴，温度 25℃"


def tool_routine(tool_call):
    """工具执行路由，agent 只调用此函数，由此函数配置具体的工具"""
    name = tool_call["function"]["name"]

    arguments = json.loads(
        tool_call["function"]["arguments"]
    )

    if name == "terminal":
        result = terminal(**arguments)

    elif name == "get_weather":
        result = get_weather(**arguments)

    else:
        result = f"Unknown tool: {name}"

    print(f"准备调用工具{name}, 其入参为 {arguments}")

    return result

agent：

model_name = 'DeepSeek-V4.1-Flash'

def agent(user_input, max_steps=20):
    # max_steps 防止无限调用

    # 首次调用 LLM 的 payload（中的 messages）：
    messages = [
        {"role": "system", "content": "You are an intelligent assistant. Use tools when necessary."},
        {"role": "user", "content": user_input}
    ]

    for _ in range(max_steps):
        payload = {
            "model": model_name,
            "messages": messages,
            # 告诉 LLM 有哪些工具可以调用：
            "tools": tools,
            "tool_choice": "auto"
        }

        resp = call_llm.call(payload)
        message = resp["choices"][0]["message"]

        # 模型的回复添加到 Context
        # 对于生产级别的 agent，这里还需要 Context 的压缩等
        messages.append(message)

        # 任务完成条件：没有 tool_calls，说明模型认为任务已经完成
        if not message.get("tool_calls"):
            return message["content"]

        # 执行工具
        for tool_call in message["tool_calls"]:
            result = tool_routine(tool_call)

            # 4. 将工具结果加入 Context
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call["id"],
                "content": result
            })

    return "Agent exceeded max_steps."

调用 agent：

answer = agent("本地文件有什么？")
answer = agent("北京天气怎么样，适合旅游吗？")


# 这个会触发多轮 工具/LLM 调用：
answer = agent("先查询所有的本地文件，然后新建一个 txt 文件，把本地文件和其摘要放进去")