基于 LLM 大语言模型 Agent 的示例

这是一个基于 LLM 的 Agent 示例，演示了如何借助 LLM Function Calling 实现 Agent Loop，并通过集成 Agent Skills 规范，实现一个支持工具调用的可扩展 Agent。

代码结构

项目结构如下：

project/
├── main.py               # 必需：HTTP 入口，提供 FastAPI app（/、/stream、/ag-ui）
├── agent/                # Agent 核心逻辑与协议适配
│   ├── core.py           # Agent Loop：流式 LLM 调用、工具执行、多轮迭代
│   ├── agui.py           # AG-UI 协议适配：Chunk 流 → AG-UI 事件流、消息格式转换
│   ├── chains.py         # 链式调用相关
│   └── prompts.py        # 提示词
├── skills.py             # Agent Skills 集成：技能发现、加载与执行
├── skills/               # 技能目录，存放各个技能
│   ├── get-current-time/
│   │   ├── SKILL.md
│   │   └── scripts/
│   │       └── now.py
│   └── create-plan/
│       └── SKILL.md
├── tools/                # 可被模型调用的工具
│   ├── get_weather.py    # 天气查询示例
│   ├── web_search.py     # 网页搜索
│   └── interface.py
├── utils/                # 通用工具与接口
│   ├── interface.py      # QueryRequest / QueryResponse 等
│   ├── formatter.py
│   └── logger.py
├── evaluation/           # 评估与测试
│   ├── run_validation.py
│   └── testcase/
│       ├── question.jsonl
│       └── question_minimal.jsonl
└── README.md

main.py 规范

main.py 是 HTTP 入口文件，必须提供一个全局的 FastAPI 对象，命名为 app。示例：

from fastapi import FastAPI
from agent.core import agent_loop
from tools.get_weather import get_weather
from utils.interface import QueryRequest, QueryResponse

app = FastAPI()

@app.post("/")
async def query(req: QueryRequest) -> QueryResponse:
    result = ""
    async for chunk in agent_loop(req.to_messages(), [get_weather]):
        if chunk.type == "tool_call" or chunk.type == "tool_call_result":
            result = ""
        elif chunk.type == "text" and chunk.content:
            result += chunk.content
    return QueryResponse(answer=result)

注意：本地启动请使用 uv run uvicorn main:app 或 uv run main.py；平台部署时会加载 main.py 中的 app 对象。

agent/core.py 与 agent/agui.py

• agent/core.py 提供 Agent Loop 核心逻辑，支持：

• 流式 LLM 调用（阿里云百炼 / DashScope 兼容 OpenAI API）

• 工具调用（Tool Calling）与执行

• 自动将工具结果回传给 LLM 继续对话

• Agent Skills 集成：自动发现 skills/ 下技能并注入系统提示，提供 load_skill_file、execute_script 等技能工具

• agent/agui.py 负责 AG-UI 协议适配：将 agent_loop 产出的 Chunk 流转换为 AG-UI 事件流（RunStarted、TextMessageStart/Content/End、ToolCallStart/Result、RunFinished 等），并支持将 AG-UI 消息列表转为 OpenAI 风格供 agent_loop 使用；/ag-ui 接口依赖此模块以 SSE 形式推送事件。

Agent Skills 技能系统

本示例集成了 Agent Skills 规范，通过技能扩展 Agent 的能力。技能是包含 SKILL.md 文件的文件夹，Agent 会自动发现并在合适的时机通过技能工具使用相应技能。

内置示例技能

技能名称	描述
get-current-time	获取当前系统时间
create-plan	创建可执行的计划

创建新技能

在 skills/ 目录下创建技能文件夹，包含 SKILL.md 文件：

---
name: my-skill
description: 技能描述，说明何时使用该技能。
---

# 技能标题

技能指令内容...

更多关于 Agent Skills 规范，请参阅 agentskills.io。

环境配置

本项目使用 uv 进行管理，可以通过 uv 运行

uv sync
uv run main.py

或使用 uvicorn 指定端口：

uv run uvicorn main:app --reload --port 7000

本示例使用阿里云百炼（DashScope）模型服务，请通过环境变量配置 API Key：DASHSCOPE_API_KEY。可在项目根目录的 .env 中配置，或由部署环境注入。

开发调试

基于 OpenAPI 的调试开发

通过平台的调试页面，可以直接使用基于 OpenAPI 规范的交互式调试界面进行开发测试：

1. 点击运行，进入 API 调试页面。
2. 在页面上填写请求参数并发送请求。
3. 实时查看响应结果与链路追踪信息。

对话调试

支持通过 AG-UI Protocol 的对话面板进行调试，可查看请求的详细信息。对话面板会请求服务的 POST /ag-ui 接口，本项目已实现该端点。

链路追踪（Trace）

平台会自动注入 instrument 收集链路追踪数据。通过对话面板或 OpenAPI 调试时，可在链路中查看请求的详细信息，包括请求耗时、LLM 调用详情、Token 使用量、错误信息等。

服务部署与调用

完成调试后，可通过服务部署将应用部署到 PAI-EAS。调用方式如下。

注意：endpoint 和 token 需在部署服务页面查看获取。

普通请求接口

Endpoint: POST /

请求体:

curl -X POST "https://<your-endpoint>/" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <your-token>" \
  -d '{"question": "北京今天的天气如何?"}'

响应体:

{
    "answer": "北京今天的天气晴朗，气温在10℃到20℃之间，空气质量良好。"
}

流式请求接口（SSE）

Endpoint: POST /stream

流式返回回答，每条 SSE 的 data 为 {"answer": "文本片段"}。请求示例：

curl -N -X POST "https://<your-endpoint>/stream" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <your-token>" \
  -d '{"question": "北京今天的天气如何?"}'

响应格式 (Server-Sent Events):

data: {"answer":"我"}
data: {"answer":"目前无法直接"}
data: {"answer":"获取实时天气信息。"}

AG-UI Protocol 接口

Endpoint: POST /ag-ui

已实现 AG-UI 协议，可对接对话面板，以 SSE 形式推送 RunStarted、TextMessageStart/Content/End、ToolCallStart/Result、RunFinished 等事件。请求示例：

curl -N -X POST "https://<your-endpoint>/ag-ui" \
  --header "Content-Type: application/json" \
  --data '{
    "messages": [
      {
        "content": "你好",
        "id": "msg_1",
        "role": "user"
      }
    ],
    "runId": "run_456",
    "threadId": "thread_123",
    "context": [],
    "tools": [],
    "forwardedProps": {},
    "state": null
  }'

响应格式 (Server-Sent Events) 示例：

data: {"type": "RUN_STARTED", "threadId": "thread_123", "runId": "run_456", "input": {...}}
data: {"type": "TEXT_MESSAGE_START", "messageId": "...", "role": "assistant"}
data: {"type": "TEXT_MESSAGE_CONTENT", "messageId": "...", "delta": "你好"}
data: {"type": "TEXT_MESSAGE_END", "messageId": "..."}
data: {"type": "RUN_FINISHED", "threadId": "thread_123", "runId": "run_456"}

评估

evaluation/ 目录下提供评估与测试用例：

• testcase/question.jsonl：问答题测试集（每行一条 {"id", "question"}）。
• run_validation.py：可用于跑验证或评估脚本（按需实现）。

依赖与运行方式见项目根目录的 pyproject.toml（建议使用 uv 管理环境）。

Agent Graph

flowchart TD
    START --> plan 
    plan --> start_hop
    start_hop --> generate_queries
    generate_queries --> search
    search --> embedding
    embedding --> extract
    extract --> route_1{route}
    route_1 -->|is relavent| commit_hop
    route_1 -->|retry times left| prepare_retry
    route_1 -->|retry times exceed| rollback
    route_1 -->|rollback times exceed| fail_and_exit
    prepare_retry --> generate_queries
    rollback --> start_hop 
    fail_and_exit --> final_answer
    commit_hop --> route_2{route}
    route_2 -->|hops left| start_hop
    route_2 --> final_answer
    final_answer --> END 

Loading

输出有格式要求 实体间建立一个图的关系，正向和反向都搜一下，这两个是可以并行的；让hop之间能够建立图的结构（DAG） 中文和英文问题区分开一下