# Aigent

**Repository Path**: ulyan/aigent

## Basic Information

- **Project Name**: Aigent
- **Description**: 一个简单易用的Python库，可以轻松地对接LLM，并可以将Python函数对象直接对接LLM，成为LLM的臂膀，实现快速搭建有特定功能的Agent。
- **Primary Language**: Python
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-05-14
- **Last Updated**: 2026-06-09

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Aigent

一个渐进式披露的 Python LLM API 库。从一行代码开始，到完全控制 —— 全部使用地道的 Python 表达。

```python
from aigent import Aigent

agent = Aigent(api_type="openai")
with agent.session() as s:
    print(s.chat("你好！"))
```

## 安装

```bash
pip install uss-aigent
```

需要 Python 3.10+。唯一外部依赖是 [httpx](https://www.pypi.org/project/httpx/)。

## 快速开始

### 第一层 —— 聊天

```python
from aigent import Aigent

# 零配置：从环境变量读取 OPENAI_API_KEY / ANTHROPIC_API_KEY
agent = Aigent(api_type="anthropic")

with agent.session(system="你是专业翻译。") as s:
    result = s.chat("翻译成英文：你好世界")
    print(result)
```

### 第二层 —— 工具

```python
from aigent import Aigent, tool

@tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """查询城市天气。"""
    # 实际代码中可调用天气 API
    return f"{city}: 22°{unit[0].upper()}"

agent = Aigent(api_type="openai")
with agent.session(tools=[get_weather]) as s:
    reply = s.chat("北京天气怎么样？")
    print(reply)
```

`@tool` 装饰器自动从函数签名和 docstring 生成 JSON Schema —— 无需手写。

想拦截工具调用？使用钩子：

```python
# Before 钩子：为所有工具调用注入用户上下文
def inject_user(args, tool_name):
    args.setdefault("user_id", current_user.id)
    return args

# After 钩子：result 是原始返回类型（dict、int、list……），不是字符串！
def log_result(args, result, error, tool_name):
    if isinstance(result, dict) and result.get("success"):
        print(f"[OK] {tool_name}: {result}")
    return result

with agent.session(tools=[get_weather]) as s:
    s.hook_all("before", inject_user)            # 作用于所有工具
    s.hook(get_weather, "after", log_result)     # 只作用于特定工具
    s.chat("北京天气怎么样？")
```

### 第三层 —— 完全控制

```python
from aigent import Aigent, Message

agent = Aigent(api_type="anthropic")

# 手动编排消息
resp = agent.raw(
    [Message.system("你是乐于助人的助手。"), Message.user("你好！")],
    max_tokens=200,
    temperature=0.7,
)
print(resp.content)   # str
print(resp.usage)     # Usage(prompt_tokens=..., completion_tokens=...)
```

## Session 与 Role API

```python
agent = Aigent()

with agent.session() as s:
    # 以用户身份聊天（默认）
    s.chat("你好")

    # 以助手身份聊天
    s.chat("我很好！", role="assistant")

    # 流式输出
    for token in s.chat("写一首诗", stream=True):
        print(token, end="")

    # 插入但不发送 —— 逐步构建上下文
    s.user.insert("我想学 Python。")
    s.assistant.insert("好选择！你想从哪里开始？")
    reply = s.chat()  # 不追加新消息，直接发送已有历史

    # 角色对象
    s.system.insert("你是一位 Python 专家。")
    s.role("tool").insert('{"result": 42}', tool_call_id="call_abc")

    # 超时控制 —— 单次请求覆盖（chat > session > agent 默认值）
    s.chat("简单问题")                      # 使用 agent/session 默认超时
    s.chat("复杂任务", timeout=180.0)       # 这次请求单独给 3 分钟

    # 工具钩子 —— 在工具调用前后拦截
    s.hook(my_tool, "before", validate_args)
    s.hook_all("after", log_all_results)

    # 历史管理
    print(s.history)   # 只读消息列表
    s.clear()           # 重置对话
```

## 工具钩子

使用 `before` 和 `after` 钩子拦截工具调用。适用于日志记录、参数注入、结果格式化、错误降级等场景。

```python
from aigent import Aigent, tool

@tool
def search(query: str) -> dict:
    """搜索知识库。"""
    return {"results": ["doc1", "doc2"], "count": 2}

@tool
def calculate(expr: str) -> float:
    """计算数学表达式。"""
    return eval(expr)

agent = Aigent(api_type="openai")

# ── 特定工具钩子 ──
with agent.session(tools=[search, calculate]) as s:
    s.hook(search, "before", lambda args, tn: {**args, "query": args["query"].strip()})
    s.hook(calculate, "after", lambda args, res, err, tn: round(res, 2) if not err else "error")
    s.chat("查找 Python 相关文档，并计算 3.14 * 2")

# ── 全局钩子（作用于所有工具）──
def log_everything(args, result, error, tool_name):
    """result 是原始返回类型 —— dict、int、list，工具返回什么就是什么。"""
    status = "FAIL" if error else "OK"
    print(f"[{status}] {tool_name}({args}) → {result}")
    return result

with agent.session(tools=[search, calculate]) as s:
    s.hook_all("after", log_everything)
    s.chat("搜索 Python 并计算 42 * 7")
```

**钩子签名：**
- `before(args: dict, tool_name: str) -> dict` — 修改/校验参数
- `after(args: dict, result: Any, error: Exception | None, tool_name: str) -> Any` — 处理结果、错误降级

**执行顺序：** 特定工具钩子先于全局钩子。`before` 链 → 工具执行 → `after` 链。

## 支持的后端

| `api_type` | 后端 | 默认模型 |
|------------|------|----------|
| `"openai"` | OpenAI Chat Completions | `gpt-4o` |
| `"anthropic"` | Anthropic Messages | `claude-sonnet-4-6` |

兼容 OpenAI 接口的服务（Azure、本地 LLM 等）可通过 `api_type="openai"` + 自定义 `base_url` 使用。

## 配置

```python
agent = Aigent(
    api_type="openai",
    api_key="sk-...",                       # 或 OPENAI_API_KEY 环境变量
    base_url="https://api.openai.com/v1",   # 或 OPENAI_BASE_URL 环境变量
    model="gpt-4o",                         # 或 OPENAI_MODEL 环境变量
    system="你是乐于助人的助手。",            # 所有 session 的默认系统提示
    timeout=30.0,                            # 浮点数（秒）或 httpx.Timeout 对象
    max_retries=3,
)
```

**超时优先级链：** `chat(timeout=...)` > `session(timeout=...)` > `agent(timeout=...)`。
timeout 也支持 `httpx.Timeout` 对象用于精细控制：

```python
from httpx import Timeout

# 分离 connect/read/write 超时 —— 适合 tool 密集场景
agent = Aigent(
    api_type="openai",
    timeout=Timeout(connect=10.0, read=120.0, write=10.0),
)

# 会话级别覆盖
with agent.session(timeout=90.0, tools=[...]) as s:
    s.chat("普通任务")                        # 使用 session 的 90s
    s.chat("繁重计算", timeout=300.0)         # 这次请求单独给 5 分钟
```

各后端环境变量：

| 变量 | OpenAI | Anthropic |
|------|--------|-----------|
| API key | `OPENAI_API_KEY` | `ANTHROPIC_API_KEY` |
| Base URL | `OPENAI_BASE_URL` | `ANTHROPIC_BASE_URL` |
| Model | `OPENAI_MODEL` | `ANTHROPIC_MODEL` |

## 错误处理

```python
from aigent import (
    Aigent,
    AuthenticationError,
    RateLimitError,
    APIError,
    ConnectionError,
    TimeoutError,
)

agent = Aigent()
try:
    with agent.session() as s:
        s.chat("你好")
except AuthenticationError:
    print("请检查 API key。")
except RateLimitError:
    print("请求过于频繁，请稍后。")
except (ConnectionError, TimeoutError):
    print("网络问题。")
except APIError as e:
    print(f"API 返回了 {e.status_code}")
```

所有异常均继承自 `LLMError`。

## 流式输出

```python
with agent.session() as s:
    for token in s.chat("给我讲个故事", stream=True):
        print(token, end="", flush=True)
    # token 会被自动累积并追加到历史中

# 流式 + 工具：真流式，不退化。
# 文本 token 实时输出，tool_calls 在后台累积。
with agent.session(tools=[get_weather]) as s:
    for token in s.chat("北京天气怎么样？", stream=True):
        print(token, end="", flush=True)
        # 用户立刻看到"让我查一下天气……"，
        # 而不是等整个 tool-call 往返完成后才出现。
```

## 许可证

Apache 2.0 —— 详见 [LICENSE](LICENSE)。