1
f(prompt) → completion

1
messages = [{"role": "user", "content": "北京今天天气怎么样？"}]
2

3
while True:
4
    response = llm(messages)                  # 1. 模型推理：要不要调用工具？
5
    if response.tool_calls:                   # 2. 决定行动
6
        for call in response.tool_calls:
7
            result = execute(call)            # 3. 执行工具（真正的"做"）
8
            messages.append(result)           # 4. 把结果塞回上下文
9
    else:
10
        return response.content               # 5. 没有工具要调，输出最终答案

1
# 把 stream=True 打开，就能拿到一个一个的事件块
2
for chunk in agent.run("用 500 字解释 TCP 三次握手", stream=True):
3
    if chunk.event == "RunContent":
4
        print(chunk.content, end="", flush=True)

1
model=OpenAIChat(
2
    id=settings.model,
3
    api_key=settings.api_key,
4
    base_url=settings.base_url,      # 指向自建网关
5
    # role_map 是给"非标准 OpenAI API"用的
6
    # 有些 API 不认 "developer" 角色，必须映射回 "system"
7
    role_map={
8
        "developer": "system",
9
        "tool": "function",
10
    },
11
)

1
from agno.tools import Toolkit
2

3
class WeatherTool(Toolkit):
4
    def __init__(self):
5
        super().__init__(name="weather")
6
        # 注册一个函数给 Agent 调用
7
        self.register(self.get_weather)
8

9
    def get_weather(self, city: str) -> str:
10
        """获取指定城市的天气"""
11
        # 实际项目里这里调用真实 API
12
        return f"{city} 今天晴天，25°C"
13

14
agent = Agent(
15
    model=OpenAIChat(id="gpt-4o"),
16
    tools=[WeatherTool()],
17
)
18

19
agent.print_response("北京今天天气怎么样？", stream=True)

1
from pathlib import Path
2
from agno.tools.workspace import Workspace
3

4
workspace_root = Path(__file__).parent
5

6
tools=[
7
    Workspace(
8
        root=workspace_root,
9
        allowed=["read", "list", "search"],   # 只读权限
10
    )
11
]

1
# 只读：让 Agent 能"看"不能"改" —— 适合代码审查、知识问答、只读分析
2
Workspace(root=workspace_root, allowed=["read", "list", "search"])
3

4
# 读写：让 Agent 能"干活" —— 适合文档生成、代码重构、批量产出
5
Workspace(root=workspace_root, allowed=["read", "list", "search", "write", "edit"])

1
用户：读一下 README.md，用三句话总结这个项目是做什么的。
2
Agent：[调用 list 找到 README.md → 调用 read 读入内容 → 总结]

1
用户：把刚才的分析结论写入 analysis.md。
2
Agent：[调用 write，在 workspace 根目录下生成 analysis.md]

1
用户：把 config.py 里的日志级别从 INFO 改成 DEBUG。
2
Agent：[read 读入 → 定位到那一行 → 局部编辑，其余保持不动]

1
用户：参照 module_template.md 的格式，为 utils/ 下每个模块生成一份说明。
2
Agent：[list 枚举模块 → 逐个套用模板 → 循环 write]

1
用户：把这份 100 页的技术规范拆成 10 个开发任务，逐个推进。
2
Agent：[read 规范 → 把任务清单写入 todo.md → 每完成一项就 edit 更新 todo.md]

1
一次 run 的生命周期                    跨 session 的沉淀
2
────────────────────                  ────────────────────
3
短期记忆（工作记忆）                    长期记忆（事实/偏好）
4
  ├─ 本轮推理的中间状态      ──写入──▶   ├─ "用户是后端工程师"
5
  └─ run 结束即消失                      ├─ "项目用的是 FastAPI"
6
                                        └─ "回答要简短、代码要带注释"
7
        ▲                                      │
8
        └──────────────读取注入────────────────┘
9
                    每轮开始时

1
# 终端多轮对话入口
2
chat = Agent(
3
    name="Chat",
4
    model=OpenAIChat(...),
5
    db=SqliteDb(db_file="chat.db"),   # 存储底座
6
    add_history_to_context=True,      # 关键：把历史带进上下文
7
    markdown=True,
8
)

1
# 第一次对话
2
agent.run("我叫张三", session_id="sess-001")
3

4
# 第二次对话，Agent 记得你叫张三
5
agent.run("我叫什么？", session_id="sess-001")  # 正确：记得
6
agent.run("我叫什么？", session_id="sess-002")  # 错误：换了会话，不记得

1
Agent(
2
    model=OpenAIChat(...),
3
    db=SqliteDb(db_file="workbench.db"),  # 长期记忆也要落在库里
4
    enable_agentic_memory=True,           # Agent 可以主动记忆对话中的关键信息
5
)

1
用户：以后回答我都希望简短一点，代码记得加注释。
2
Agent：[识别到这是稳定的用户偏好 → 写入长期记忆]
3

4
用户：我们项目是 FastAPI + SQLAlchemy。
5
Agent：[识别到这是项目事实 → 写入长期记忆]

1
[新一轮对话开始]
2
Agent：[检索长期记忆 → 发现"偏好简短、代码带注释""项目用 FastAPI"]
3
      → 本次回答自动遵循这些约束

1
agno-agent/
2
├── config.py          # 配置管理（从 .env 读取）
3
├── agent.py           # Agent 定义，可复用
4
├── chat.py            # 终端多轮对话入口
5
├── server.py          # FastAPI 后端，提供 SSE 流式接口
6
├── frontend/          # React 前端
7
│   ├── package.json
8
│   ├── App.tsx
9
│   └── ...
10
└── .env               # API 密钥等敏感配置

1
import os
2
from dataclasses import dataclass
3
from pathlib import Path
4

5
from dotenv import load_dotenv
6

7
# 自动加载 .env，不管从哪个目录启动
8
load_dotenv(Path(__file__).resolve().parent / ".env")
9

10

11
class ConfigurationError(RuntimeError):
12
    """配置缺失时的自定义异常"""
13

14

15
@dataclass(frozen=True)
16
class Settings:
17
    """不可变配置对象，防止运行时被意外修改"""
18

19
    base_url: str
20
    api_key: str
21
    model: str
22

23

24
def load_settings() -> Settings:
25
    api_key = os.getenv("OPENAI_API_KEY", "").strip()
26
    if not api_key:
27
        raise ConfigurationError(
28
            "OPENAI_API_KEY 是必填项，请在 .env 或环境变量中设置"
29
        )
30
    return Settings(
31
        base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1").strip(),
32
        api_key=api_key,
33
        model=os.getenv("OPENAI_MODEL", "gpt-4o").strip(),
34
    )

1
import os
2
from pathlib import Path
3

4
from agno.agent import Agent
5
from agno.db.sqlite import SqliteDb
6
from agno.models.openai import OpenAIChat
7
from agno.tools.workspace import Workspace
8

9
from config import load_settings
10

11

12
def create_agent() -> Agent:
13
    settings = load_settings()
14
    workspace_root = Path(__file__).parent
15
    return Agent(
16
        name="Workbench",
17
        model=OpenAIChat(                 # ← Model：推理能力
18
            id=settings.model,
19
            api_key=settings.api_key,
20
            base_url=settings.base_url,
21
            # role_map 是给"非标准 OpenAI API"用的
22
            # 有些 API 不认 "developer" 角色，必须映射回 "system"
23
            role_map={
24
                "developer": "system",
25
                "tool": "function",
26
            },
27
        ),
28
        tools=[
29
            Workspace(                    # ← Workspace：工作空间能力
30
                root=workspace_root,
31
                allowed=["read", "list", "search"],  # 只读权限
32
            )
33
        ],
34
        db=SqliteDb(db_file="workbench.db"),  # ← Memory：会话持久化到 SQLite
35
        enable_agentic_memory=True,           # ← Memory：Agent 主动记忆关键信息
36
        add_history_to_context=True,          # ← Memory：每轮自动带上历史
37
        markdown=True,
38
    )
39

40

41
# 模块级单例，供 server.py 和 main.py 复用
42
workbench = create_agent() if os.getenv("OPENAI_API_KEY", "").strip() else None

1
# chat.py —— 命令行多轮对话 Agent
2
from agno.agent import Agent
3
from agno.db.sqlite import SqliteDb
4
from agno.models.openai import OpenAIChat
5

6
from config import load_settings
7

8
# 固定会话 ID，让多轮对话落在同一个会话里
9
SESSION_ID = "default"
10

11

12
def main() -> None:
13
    s = load_settings()
14
    chat = Agent(
15
        name="Chat",
16
        model=OpenAIChat(
17
            id=s.model,
18
            api_key=s.api_key,
19
            base_url=s.base_url,
20
            # 覆盖默认的 system→developer 映射
21
            role_map={
22
                "system": "system",
23
                "user": "user",
24
                "assistant": "assistant",
25
                "tool": "tool",
26
                "model": "assistant",
27
            },
28
        ),
29
        db=SqliteDb(db_file="chat.db"),
30
        add_history_to_context=True,  # 关键：把历史带进上下文
31
        markdown=True,
32
    )
33

34
    print("开始对话（输入 exit / quit 退出）\n")
35
    while True:
36
        try:
37
            user_input = input("你：").strip()
38
        except (EOFError, KeyboardInterrupt):
39
            print("\n再见！")
40
            break
41

42
        if user_input.lower() in {"exit", "quit", "q"}:
43
            print("再见！")
44
            break
45
        if not user_input:
46
            continue
47

48
        # 每次传同一个 session_id，历史才能连续
49
        chat.print_response(user_input, stream=True, session_id=SESSION_ID)
50
        print()
51

52

53
if __name__ == "__main__":
54
    main()

1
import json
2
import asyncio
3

4
from fastapi import FastAPI
5
from fastapi.responses import StreamingResponse
6
from fastapi.middleware.cors import CORSMiddleware
7
from pydantic import BaseModel
8

9
from agent import workbench
10

11
app = FastAPI(title="Agno Agent API")
12

13
# 允许跨域，前端 localhost 调后端时必配
14
app.add_middleware(
15
    CORSMiddleware,
16
    allow_origins=["*"],
17
    allow_methods=["*"],
18
    allow_headers=["*"],
19
)
20

21

22
class ChatRequest(BaseModel):
23
    message: str
24
    session_id: str = "default"
25

26

27
async def event_stream(message: str, session_id: str):
28
    """
29
    把 Agno 的流式事件包装成 SSE 格式。
30
    SSE 协议非常朴素：
31
      data: 内容\n\n
32
    每行以 "data: " 开头，两个换行结尾。
33
    """
34
    try:
35
        # 把 Agno 的同步迭代器拿到线程池里跑
36
        # 不然会阻塞 FastAPI 的事件循环
37
        iterator = await asyncio.to_thread(
38
            lambda: workbench.run(
39
                message,
40
                stream=True,
41
                session_id=session_id,
42
            )
43
        )
44

45
        for chunk in iterator:
46
            event_type = chunk.event
47

48
            if event_type == "RunStarted":
49
                # 通知前端：开始接收
50
                yield f"data: {json.dumps({'type': 'start'})}\n\n"
51

52
            elif event_type == "RunContent":
53
                # 这是真正的内容增量，一句一句往外吐
54
                if chunk.content:
55
                    yield f"data: {json.dumps({'type': 'content', 'content': chunk.content})}\n\n"
56

57
            elif event_type == "ToolCallStarted":
58
                # Agent 在调用工具，通知前端显示状态
59
                tool_name = chunk.tools[0].tool_name if chunk.tools else "unknown"
60
                yield f"data: {json.dumps({'type': 'tool_start', 'tool': tool_name})}\n\n"
61

62
            elif event_type == "ToolCallCompleted":
63
                yield f"data: {json.dumps({'type': 'tool_end'})}\n\n"
64

65
            elif event_type == "RunCompleted":
66
                yield f"data: {json.dumps({'type': 'done'})}\n\n"
67

68
            elif event_type == "RunError":
69
                error_msg = chunk.content or "Unknown error"
70
                yield f"data: {json.dumps({'type': 'error', 'content': error_msg})}\n\n"
71

72
    except Exception as e:
73
        yield f"data: {json.dumps({'type': 'error', 'content': str(e)})}\n\n"
74

75

76
@app.post("/chat")
77
async def chat_endpoint(req: ChatRequest):
78
    """
79
    前端 POST /chat，后端返回 StreamingResponse，
80
    浏览器用 EventSource 或 fetch 逐行读取。
81
    """
82
    return StreamingResponse(
83
        event_stream(req.message, req.session_id),
84
        media_type="text/event-stream",
85
        headers={
86
            "Cache-Control": "no-cache",
87
            "Connection": "keep-alive",
88
            "X-Accel-Buffering": "no",  # 禁用 Nginx 缓冲（生产环境部署时要加）
89
        },
90
    )
91

92

93
@app.get("/health")
94
async def health():
95
    return {"status": "ok", "agent": workbench.name if workbench else None}
96

97

98
if __name__ == "__main__":
99
    import uvicorn
100
    uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True)

1
.cursor {
2
  animation: blink 1s step-end infinite;
3
}
4

5
@keyframes blink {
6
  50% {
7
    opacity: 0;
8
  }
9
}

1
role_map = {
2
    "developer": "system",  # 把 developer 映射回 system
3
    "tool": "function",     # 工具调用也映射
4
}

1
iterator = await asyncio.to_thread(
2
    lambda: workbench.run(message, stream=True, session_id=session_id)
3
)

1
# 第一次对话
2
agent.run("我叫张三", session_id="sess-001")
3

4
# 第二次对话，Agent 记得你叫张三
5
agent.run("我叫什么？", session_id="sess-001")  # 正确：记得
6
agent.run("我叫什么？", session_id="sess-002")  # 错误：不记得

1
const TIMEOUT_MS = 60000; // 60 秒超时
2
const timeoutId = setTimeout(() => {
3
  abortController.abort();
4
  setMessages(prev => [...prev, {
5
    role: "assistant" as const,
6
    content: "⏱️ 请求超时，请重试",
7
  }]);
8
}, TIMEOUT_MS);
9

10
// 在 finally 中清除定时器

1
Agent
2
├── Model（推理能力）                    —— 想
3
├── Tools（外部行动能力）                 —— 做
4
├── File System / Workspace（工作空间能力） —— 存
5
├── Memory（长期状态能力）                —— 记
6
└── Runtime Environment（执行环境）        —— 跑

1
用户输入 → React 前端 → POST /chat (SSE) → FastAPI 后端（Runtime）
2
    → agent.run(stream=True) → Agno Agent
3
        ├── OpenAIChat     → 模型 API          （Model）
4
        ├── Workspace      → 读写本地文件       （File System）
5
        ├── SqliteDb       → 消息与记忆持久化    （Memory）
6
        └── 工具调用（可选）→ 执行外部动作        （Tools）
7
    → RunContentEvent 迭代器
8
    → 包装成 SSE "data: " 格式
9
    → 前端逐行解码 → 逐字渲染