平台架构:API 服务、查询层与 Worker
30 秒导读: 前面几章讲的是 Julep 的业务(任务怎么跑、步骤怎么求值、工具怎么调、记忆怎么检索)。这一章只讲工程装配:一个 HTTP 请求在进程层面怎么 从网关走到数据库,一个后台任务怎么在另一个进程里被 Temporal 拉起。看完你能画出 Julep 的"运行时拓扑图",知道每个容器是干什么的、代码在哪个文件装配起来。
本章不重复业务逻辑。任务状态机看 02,步骤与表达式看 03,工具集成看 04,混合检索看 05,核心数据模型看 01。
1. 这是什么(零基础也能懂)
一句话定义: Julep 的"平台层"是把一堆现成的开源组件——FastAPI、asyncpg、Temporal、Traefik、LiteLLM、S3——用胶水代码接成一台机器,让 agent 的业务代码有地方跑。
它要解决的问题: agent 平台有两种截然不同的活儿。
- 要立刻回应的活:创建 agent、查任务、发一条聊天——用户在等,必须同步、快、几百毫秒内返回。
- 要跑很久的活:一个任务工作流可能跑几分钟到几天,中间调 LLM、调工具、等外部 API——不能占着一个 HTTP 连接干等。
Julep 把这两类活拆到两种进程里:
| 进程 | 干什么 | 怎么起 |
|---|---|---|
API 服务(web:app) | 处理同步 HTTP 请求, 读写 Postgres | gunicorn 拉起 FastAPI |
Worker(agents_api.worker) | 处理长时后台任务,跑 Temporal 工作流 | python -m agents_api.worker |
关键点:两种进程是同一个代码库、同一个 Docker 镜像基底,只是入口不同。 API 进程入口是 agents_api.web:app,Worker 进程入口是 agents_api.worker——见两个 Dockerfile 的 ENTRYPOINT(Dockerfile:33、Dockerfile.worker:33)。
一句话直觉: 把 API 服务当"前台接待"(来一个客人接待一个,快进快出),把 Worker 当"后厨"(接了长订单慢慢做,做好了把结果写回台账)。台账就是 Postgres,后厨的排期系统就是 Temporal。
2. 顶层全景(它大概怎么转)
2.1 服务拓扑
docker-compose.yml 用 include: 把十来个子服务的 compose 文件拼在一起(docker-compose.yml:6-19)。下面这张图是一个请求 / 一个任务在这些容器间的走向——从左到右是同步请求链,右下是异步任务链:
同步请求链(用户在等)
┌────────┐ /api/* ┌──────────┐ ┌────────────┐
│ 用户 │ ─────────▶ │ gateway │ 注入头部 │ agents-api │
│ (JWT) │ │ (Traefik) │ ─────────▶ │ (FastAPI/ │
└────────┘ │ JWT 校验 │ X-Dev-Id │ gunicorn) │
└──────────┘ └─────┬──────┘
│ asyncpg 连接池
▼
┌──────────────────┐
│ memory-store (PG │
│ + TimescaleDB + │
│ pgvector) │
└──────────────────┘
─────────────────────────────────────────────────────────────
异步任务链(后台慢慢跑)
agents-api ──start_workflow──▶ ┌───────────┐ ──派活──▶ ┌────────┐
(create execution) │ scheduler │ │ worker │
│ (Temporal)│ ◀─心跳/结果─│ 进程 │
└───────────┘ └───┬────┘
│ 调用
┌──────────────┬───────────────┼──────────────┐
▼ ▼ ▼ ▼
┌──────────┐ ┌────────────┐ ┌───────────┐ ┌──────────┐
│ llm-proxy│ │integrations│ │ blob-store│ │ memory- │
│(LiteLLM) │ │ -service │ │(S3/Seaweed)│ │ store │
└──────────┘ └────────────┘ └───────────┘ └──────────┘
2.2 部件一句话职责
| 部件 | 干什么 | 在哪 |
|---|---|---|
| gateway | Traefik 反向代理:路由、JWT 校验、注入 X-Developer-Id | src/gateway/traefik.yml.template |
| agents-api | FastAPI 应用,同步 REST 层 | agents_api/web.py、app.py |
| worker | Temporal worker 进程,跑工作流/活动 | agents_api/worker/ |
| scheduler | Temporal 服务端(auto-setup)+ UI | src/scheduler/docker-compose.yml |
| memory-store | Postgres(TimescaleDB + pgvector),主数据库 | src/memory-store/ |
| llm-proxy | LiteLLM 代理,统一各家 LLM | src/llm-proxy/ |
| integrations-service | 工具/集成执行器,HTTP 调用 | src/integrations-service/ |
| blob-store | S3 兼容对象存储(SeaweedFS),存大 payload | src/blob-store/ |
注意 API 服务有两副面孔(agents-api/docker-compose.yml:100-114,由 compose profiles 切换):
agents-api(单租户):自己拿AGENTS_API_KEY鉴权。agents-api-multi-tenant:信任 gateway 已校验 JWT,只认头里的X-Developer-Id,并把AGENTS_API_PREFIX设成/api。
两种模式的鉴权分工写在顶层 compose 的注释里(docker-compose.yml:24-54)。
3. 核心原理一:FastAPI 应用怎么装配
这节讲同步请求那半边是怎么搭起来的。三个文件分工明确:
| 文件 | 职责 |
|---|---|
app.py | 造 FastAPI 实例 + lifespan(建/拆连接池、S3) |
web.py | 挂路由、中间件、异常处理器、进程入口 |
routers/ | 各资源子路由,层层聚合 |
3.1 lifespan:进程启动时建连接池
FastAPI 的 lifespan 是一个异步上下文管理器——进程起来时跑一遍 yield 前的代码(建资源),进程关掉时跑 yield 后的代码(拆资源)。Julep 在这里建两样长期资源:Postgres 连接池和 S3 客户端,都挂在 app.state 上。
# 示意,非源码:lifespan 的骨架
@asynccontextmanager
async def lifespan(container):
pool = await create_db_pool(pg_dsn, max_size=..., min_size=...) # 建 PG 连接池
container.state.postgres_pool = pool # 挂到 app.state
session = get_session()
container.state.s3_client = await session.create_client("s3", ...) # 建 S3 客户端
try:
yield # ← 进程在这里服务请求
finally:
await pool.close() # 关闭连接池
await s3_client.close() # 关闭 S3
真实实现见 app.py:30-68(lifespan)。几个值得注意的细节:
- 连接池由
create_db_pool建(clients/pg.py:18create_db_pool),max_size取自pool_max_size——被min(..., 10)硬顶到最多 10(env.py:79)。 min_size=min(pool_max_size, 10)(app.py:34):池子一起来就预热到满,避免冷启动打洞。- Worker 进程复用同一个
lifespan:worker/__main__.py:41用async with lifespan(app):把 worker 包在里面——所以 worker 也拿到同一套连接池和 S3 客户端,同步查询层的代码在 worker 里照样能用。
3.2 app 实例与 metrics
app.py:71-92 造 FastAPI 实例,把 lifespan 传进去,root_path=api_prefix(多租户下是 /api),并用 Instrumentator(prometheus)自动埋点、暴露 /metrics。文档页走 Scalar(app.py:98-108)。
3.3 web.py:挂路由 + 中间件 + 异常处理器
web.py 是"把 app 武装起来"的地方,做四件事:
① 聚合子路由。 每个资源一个 router,统一挂 get_api_key 依赖鉴权(web.py:278-288):
# 摘自 web.py:278-288(节选)
app.include_router(agents.router, dependencies=[Depends(get_api_key)])
app.include_router(sessions.router, dependencies=[Depends(get_api_key)])
app.include_router(tasks.router, dependencies=[Depends(get_api_key)])
# ... users / files / docs / secrets / projects
app.include_router(internal.router) # 内部路由不加鉴权依赖
enable_responses 开关决定挂 responses 单路由还是挂全套业务路由(web.py:275-286)。
② 用量中间件。 usage_check_middleware(web.py:293)在每个请求前查开发者的累计花费,超免费额度就 403 拦 截——GET 和带 paid 标签的用户放行。
③ 异常处理器工厂。 make_exception_handler(web.py:57)把异常统一转成 {"error": {...}} 的 JSON;web.py:386-470 为 HTTPException、Temporal RPCError、BaseCommonException、LiteLLM 的 BadRequestError/APIError 各注册一个处理器,把底层异常翻译成带 code/fix 的友好响应。
④ 进程入口。 main()(web.py:473)用 uvicorn 起服务;文件末尾 asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())(web.py:492)换成 uvloop 提速。生产上真正的入口是 gunicorn:ENTRYPOINT ... gunicorn agents_api.web:app -c gunicorn_conf.py(Dockerfile:33),worker class 用 uvicorn.workers.UvicornWorker、绑 0.0.0.0:8080(gunicorn_conf.py:8-9)。
3.4 routers/ 的三层聚合
路由目录是逐层 re-export的:
routers/__init__.py ← 顶层聚合:from .tasks import router as tasks_router
└─ routers/tasks/__init__.py ← 子聚 合:from .router import router
├─ router.py ← 定义 APIRouter()
├─ get_task_details.py ← 一个 endpoint 一个文件
├─ create_task.py
└─ ...
一个资源(如 tasks)的每个 endpoint 拆成独立文件,由该资源的 __init__.py 汇总(routers/tasks/__init__.py:1-13),再由顶层 routers/__init__.py 统一导出给 web.py。好处是每个 handler 文件小、可单独 grep 定位。
4. 核心原理二:查询层——把"写 SQL 的函数"包成安全的 asyncpg 调用
这是 Julep 工程上最有辨识度的模式。约定:一个查询函数只负责返回 (SQL 字符串, 参数列表, 取数方式) 这个元组,真正的执行、连接管理、异常翻译、结果包成 Pydantic 模型,全交给装饰器。
4.1 为什么这么做
裸写 asyncpg 每个查询都要重复:从池子借连接、开事务、await conn.fetch(...)、catch 各种 asyncpg.*Error 再转成 HTTP 码、把 Record 转成模型。Julep 把这些横切关注点抽成一摞装饰器,查询函数本体只剩"业务 SQL + 参数"。
4.2 三层装饰器叠罗汉
以 queries/tasks/get_task.py 为范例,函数头上叠了三个装饰器(get_task.py:56-64),从下往上是执行顺序:
# 摘自 get_task.py:56-64
@rewrap_exceptions(common_db_exceptions("task", ["get"])) # ③ 把 DB 异常翻译成 HTTP
@wrap_in_class(spec_to_task, one=True) # ② 把 Record 包成 Task 模型
@pg_query # ① 拿元组去真正跑 SQL
@beartype
async def get_task(*, developer_id: UUID, task_id: UUID):
return (get_task_query, [developer_id, task_id], "fetchrow") # 只返回元组
数据流:
调用 get_task(developer_id=..., task_id=...)
│
▼ 函数体返回 (SQL, [参数], "fetchrow")
[① @pg_query] 从 app.state.postgres_pool 借连接、开事务、执行 → list[Record]
│
▼
[② @wrap_in_class] 每个 Record → spec_to_task(**record) → Task 模型
│
▼
[③ @rewrap_exceptions] 若中途抛 asyncpg 异常 → 映射成对应的 HTTPException
│
▼ 返回 Task(或抛 404/409/...)
4.3 @pg_query:执行引擎
pg_query(queries/utils.py:91 pg_query)是核心。 被它包住的函数返回值先经 prepare_pg_query_args(utils.py:58)规整成 (fetch方法名, {query, args, timeout}) 的批次,然后:
# 示意,非源码:pg_query wrapper 的核心
pool = connection_pool or app.state.postgres_pool # 默认用 lifespan 建的池
async with pool.acquire() as conn, conn.transaction(): # 借连接 + 开事务
for method_name, payload in batch: # 支持一批多条
method = getattr(conn, method_name) # conn.fetch / fetchrow / fetchmany
args = sanitize_string(payload["args"]) # 剔除 � 空字符
results = await method(payload["query"], *args, timeout=payload.get("timeout"))
真实实现见 utils.py:112-182。几个精华细节:
- 超时统一由
query_timeout注入(env.py:78,默认 90s),在prepare_pg_query_args里塞进每条(utils.py:69)。 fetchrow空结果被显式转成NoDataFoundError(utils.py:149-151),好让上层装饰器映射成 404。sanitize_string(utils.py:197)递归剥掉字符串里的�——Postgres 的 text 类型不接受 NUL 字节,这是个真实的坑。- socket 错误特判:连不上 DB(
socket.gaierror)直接抛 429 "Resource busy"(utils.py:164-174)。 return_index(默认-1)决定一批查询里返回哪条的结果(utils.py:179)。
配套的 wrap_in_class(utils.py:219)把 Record 列表转成 Pydantic 模型,one=True 时断言恰好一条、maybe_one=True 时允许零或一条。
4.4 rewrap_exceptions + common_db_exceptions:把数据库错翻译成 HTTP 码
rewrap_exceptions(utils.py:288)接一个 {异常判据 → 转换器} 的映射:catch 到匹配的异常就换成映射里的新异常抛出(并挂上 __cause__ 保留原始栈)。判据可以是异常类型,也可以是 lambda e: ... 谓词(utils.py:295-309)。
映射本体不用每个查询手写,而是由 common_db_exceptions(resource_name, operations) 工厂统一生成(common/utils/db_exceptions.py:33 common_db_exceptions)。它把一大票 asyncpg 异常映射到合适的 HTTP 状态码:
| 数据库异常 | → HTTP | 含义 |
|---|---|---|
ForeignKeyViolationError | 404 | 引用的资源不存在 |
UniqueViolationError | 409 | 唯一键冲突 |
CheckViolationError | 400 | 违反约束 |
NoDataFoundError | 404 | 没查到 |
NotNullViolationError | 400 | 必填字段为空 |
socket.gaierror | 429 | 连接不上,稍后重试 |
StringDataRightTruncationError | 400 | 文本超长 |
见 db_exceptions.py:69-215。它还会按 operations 追加操作特定规则:delete 时把"仍被引用"的外键冲突转成 409、update 时把基数冲突转成 409(db_exceptions.py:218-238)。所以 common_db_exceptions("task", ["get"]) 生成的映射,专门为"get task"这个操作定制了报错文案(get_operation_message,db_exceptions.py:54-58)。
可借鉴的精华: 查询函数瘦到只剩 SQL + 参数,一切样板(连接、事务、超时、NUL 清洗、异常翻译、模型包装)由声明式装饰器栈接管。加一个查询 = 写一段 SQL + 叠三个装饰器,错误处理"零成本"复用。
5. 核心原理三:客户端层 clients/——对外部服务的统一封装
clients/ 把每个外部依赖包成一个薄模块,应用其他地方只 import 函数、不碰连接细节。
| 客户端 | 封装谁 | 关键函数 / 说明 |
|---|---|---|
pg.py | Postgres | create_db_pool——建 asyncpg 池,_init_conn 给每条连接注册 json/jsonb 编解码器(pg.py:8-23) |
temporal.py | Temporal | get_client / run_task_execution_workflow——连服务端、起工作流 |
litellm.py | LiteLLM 代理 | acompletion / aembedding——统一 LLM 调用,支持用户自带 key |
async_s3.py | blob-store | add_object_with_hash / get_object——内容寻址存取 |
integrations.py | integrations-service | run_integration_service——HTTP POST 到 /execute/{provider} |
几处值得点出的设计:
async_s3用内容哈希做 key。add_object_with_hash(async_s3.py:87)用xxh3_64_hexdigest(body)当对象 key——同样内容天然去重,add_object里先exists再写(async_s3.py:58-69)。S3 客户端从app.state.s3_client取(即 lifespan 建的那个),setup()用@alru_cache(maxsize=1)保证只初始化一次并确认 bucket 存在(async_s3.py:13-34)。temporal.get_client三种鉴权。 mTLS(证书)、API key、或都不配(本地明文),按环境变量分支(temporal.py:41-58)。get_client_with_metrics(temporal.py:61)额外挂一个 Prometheus runtime——worker 专用这个带指标的客户端。integrations是纯 HTTP 客户端。 用 httpxAsyncClientPOST 到 integrations-service,600s 超时(integrations.py:24-31)。工具真正在那个独立服务里执行,agents-api 只发请求。
6. 核心原理四:Temporal Worker——后台任务那半边
这节讲异步任务链的进程装配。业务上的工作流/状态机在 02;这里只讲"worker 进程怎么起、注册了什么、payload 怎么编解码"。
6.1 worker.py:注册 workflow 与 activities
create_worker(worker/worker.py:16)造一个 Temporal Worker,把工作流类和活动函数注册进去、绑到任务队列上:
# 示意,非源码:worker 注册骨架
Worker(
client,
task_queue=temporal_task_queue, # 默认 "julep-task-queue"
workflows=[DemoWorkflow, TaskExecutionWorkflow],
activities=[*task_activities, # task_steps 里所有函数,反射自动收集
execute_integration, execute_system, execute_api_call,
save_inputs_remote, load_inputs_remote, pg_query_step],
max_concurrent_workflow_tasks=..., # 并发上限来自 env
max_concurrent_activities=...,
)
真实实现见 worker.py:39-62。活动是反射收集的:getmembers(task_steps, isfunction) 把 task_steps 模块里所有函数一次性抓成活动列表(worker.py:36),不用手写每一个。并发/限速四个参数全来自环境变量(worker.py:8-13、58-61)。任务队列名 julep-task-queue(env.py:125)——API 侧起工作流时也用同一个队列(temporal.py:132),两边靠这个名字对上。
6.2 main.py:进程入口 + 自愈重试
worker/__main__.py:30 的 main() 是 worker 容器的入口(Dockerfile.worker:33 的 ENTRYPOINT 指向 python -m agents_api.worker):
# 示意,非源码:worker 入口
@retry(wait=wait_fixed(20), retry=retry_if_exception_type(RuntimeError)) # 连不上就每 20s 重试
async def main():
client = await temporal.get_client_with_metrics() # 带 Prometheus 的客户端
worker = create_worker(client)
async with lifespan(app): # ← 复用 API 的 lifespan,拿到同一套 PG 池 + S3
await worker.run()
见 __main__.py:25-43。两个要点:tenacity 的 @retry 让 worker 在 Temporal 还没起来时死等重连;async with lifespan(app) 让 worker 共享 §3.1 那套资源初始化——这就是为什么 worker 里能直接用查询层。
6.3 codec.py:payload 编解码 + 大对象走 blob-store
Temporal 在客户端和 worker 之间传的所有数据都要序列化,而且单个 payload 有大小上限。Julep 自定义了一套 codec 解决两件事:让 Pydantic 模型能传、让超大对象绕过 Temporal 走 S3。
① 自定义序列化格式。 PydanticEncodingPayloadConverter(codec.py:167)用编码 text/pickle+lz4:先 larch.pickle 序列化、再 lz4 压缩(serialize,codec.py:54),并在 payload metadata 里记 Python 版本以防跨版本反序列化(codec.py:200-211)。它被塞进 PydanticPayloadConverter(codec.py:235)排在默认转换器前面,最终组装成 pydantic_data_converter(codec.py:245)供客户端和 worker 共用。
② 大对象 offload。 这是关键机制。RemoteObject(codec.py:32)是一个"指针":只存 {key, bucket},真身在 blob-store 里。
payload 要过 Temporal
│
▼
total_size(obj) > blob_store_cutoff_kb(默认 64KB)?
│ 否 │ 是
▼ ▼
原样序列化传过去 RemoteObject.from_value(obj)
├─ serialize(obj) → 压缩字节
├─ 存进 S3,key = 内容哈希
└─ 只把 {key,bucket} 这个小指针传过去
│
worker 收到后 .load() 从 S3 取回真身
判大小 + 换指针的逻辑在 offload_if_large(common/interceptors.py:103):超过 blob_store_cutoff_kb(env.py:64,默认 64KB)就 RemoteObject.from_value(result)(interceptors.py:113-114);反向的 load_if_remote(interceptors.py:81)在 worker 端把 RemoteObject 换回真身。整个开关由 use_blob_store_for_temporal(env.py:58)控制。
③ 两个专门的 activity 做批量 offload。 save_inputs_remote / load_inputs_remote(activities/sync_items_remote.py:23-24)把一整批输入分别过一遍 offload_if_large / load_if_remote——工作流里显式调它们,把大列表整体挪到 blob-store。API 侧起工作流前也会先 offload_if_large(arguments)(temporal.py:123-124),避免启动 payload 就爆。
拦截器 offload_to_blob_store(interceptors.py:119)则把这套 load/offload 织进每个 activity/workflow 的调用边界,对业务代码透明。
可借鉴的精华: 用"内容寻址的 RemoteObject 指针 + 大小阈值"把工作流引擎的 payload 限制透明化——业务代码照常传大对象,超阈值的自动进 S3、留个小指针,worker 用时按需拉回。去重还是白送的(哈希 key)。
7. 服务拓扑怎么接起来(落地)
前面各部件,靠 docker-compose.yml 的 include: 组装(docker-compose.yml:6-19)。这节把"请求/任务怎么穿过这些容器"收口。
7.1 gateway:Traefik 做的三件事
gateway 是一个定制的 Traefik 镜像(带自研 JWT 插件),配置由 traefik.yml.template 渲染。它按路径把请求分发到后端(traefik.yml.template:19-155):
| 路径 | 去哪 | 备注 |
|---|---|---|
/api/temporal/{path} | agents-api | 内部 temporal 端点,priority 2 |
/api/{swagger|docs|openapi.json} | agents-api | 文档,免鉴权 |
/api/healthz | agents-api | 健康检查 |
/api/* | agents-api | 主 API,过 JWT 鉴权链 |
/tasks-ui/* | temporal-ui | Temporal 控制台 |
/v1/graphql | hasura | GraphQL |
/analytics/* | analytics | Metabase(仅 iframe) |
JWT 鉴权链是核心中间件(traefik.yml.template:174-200):/api/* 先过 api-jwt-auth(用自研 jwt 插件校验 HS512 签名,把 JWT 的 sub claim 提取成 X-Developer-Id 头),再过 agents-api-add-headers(注入内部 AGENTS_API_KEY),最后 strip-prefix 去掉 /api。所以到达 agents-api 的请求已经带好了可信的 X-Developer-Id——这正是多租户模式期望的输入。自研插件声明在 experimental.localPlugins(traefik.yml.template:254-257)。
7.2 各支撑服务
- scheduler = Temporal。
temporalio/auto-setup镜像 + 自己的 Postgres(temporal-db)+ 两个 UI(src/scheduler/docker-compose.yml:37-92)。worker 和 API 都连它。 - llm-proxy = LiteLLM。
berriai/litellm-database镜像,自带 Postgres + Redis,配置挂litellm-config.yaml(src/llm-proxy/docker-compose.yml:3-15)。agents-api 和 worker 都把这份 config 只读挂进去(agents-api/docker-compose.yml:98-99)。 - blob-store = SeaweedFS。 HA 版是一个 master + 多个 volume 节点组成的 S3 兼容集群(
src/blob-store/docker-compose-ha.yml:9-63)。§6.3 的 RemoteObject 就存这里。 - memory-store = Postgres 系。 agents-api 查询层(§4)打的就是它。
7.3 一次请求的全链(把各节串起来)
用户带 JWT
│ POST /api/tasks/{id}/executions
▼
gateway(Traefik):校验 JWT → 注入 X-Developer-Id + 内部 API key → strip /api
▼
agents-api(gunicorn→FastAPI):usage 中间件查额度 → get_api_key 依赖鉴权
▼ 路由到 create_task_execution handler
查询层(@pg_query 栈):写 executions 行 → memory-store(PG)
▼
clients/temporal.run_task_execution_workflow:offload 大参数 → start_workflow
▼ 返回 202 给用户(同步链到此结束)
─────────────────── 异步链接管 ───────────────────
scheduler(Temporal)把工作流派到 julep-task-queue
▼
worker 进程:codec 解 payload(必要时从 blob-store load)→ 跑 TaskExecutionWorkflow
▼ 活动里调 llm-proxy / integrations-service,状态转移写回 memory-store
用户后续 GET /api/.../executions 轮询结果
8. 边界与局限(诚实)
- docker-compose 不是生产级。 顶层文件开头就写着
TODO: Make docker compose setup production ready(docker-compose.yml:1-2)。 - 连接池被硬顶到 10。
pool_max_size无论 CPU 多少都min(..., 10)(env.py:79),高并发下可能是瓶颈——但配合 gunicorn 多 worker 进程,每进程各有一池。 - codec 很脆。
codec.py:1-5自己都写着"Working with temporal's codec is really really weird ... quite brittle"。pickle + Python 版本绑定意味着跨 Python 版本或改数据结构容易踩坑。 - content-length 校验被注释掉了。
app.py:111-126有一段限制 payload 大小的中间件,目前是死代码。 - 鉴权模式是编译期分叉。 单/多租户靠 compose profile + 环境变量切,不是运行时可切换。
9. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| FastAPI 实例 + lifespan | agents-api/agents_api/app.py:30,71 | lifespan、app |
| 建 PG 连接池 | agents-api/agents_api/clients/pg.py:18 | create_db_pool |
| 挂路由/中间件/异常/入口 | agents-api/agents_api/web.py | register_exceptions、usage_check_middleware、main |
| 路由聚合 | agents-api/agents_api/routers/__init__.py、routers/tasks/__init__.py | (re-export) |
| 查询执行装饰器 | agents-api/agents_api/queries/utils.py:91,219,288 | pg_query、wrap_in_class、rewrap_exceptions |
| 查询范例 | agents-api/agents_api/queries/tasks/get_task.py:60 | get_task |
| DB 异常映射工厂 | agents-api/agents_api/common/utils/db_exceptions.py:33 | common_db_exceptions |
| Temporal 客户端 / 起工作流 | agents-api/agents_api/clients/temporal.py:33,99 | get_client、run_task_execution_workflow |
| S3 内容寻址存取 | agents-api/agents_api/clients/async_s3.py:87 | add_object_with_hash |
| LLM 调用封装 | agents-api/agents_api/clients/litellm.py:57 | acompletion |
| 集成服务 HTTP 客户端 | agents-api/agents_api/clients/integrations.py:12 | run_integration_service |
| worker 注册 | agents-api/agents_api/worker/worker.py:16 | create_worker |
| worker 入口 + 重试 | agents-api/agents_api/worker/__main__.py:30 | main |
| Temporal codec + RemoteObject | agents-api/agents_api/worker/codec.py:32,235,245 | RemoteObject、PydanticPayloadConverter、pydantic_data_converter |
| 大对象 offload 判定 | agents-api/agents_api/common/interceptors.py:103,119 | offload_if_large、offload_to_blob_store |
| 批量 offload 活动 | agents-api/agents_api/activities/sync_items_remote.py:23 | save_inputs_remote、load_inputs_remote |
| Traefik 路由 + JWT 链 | src/gateway/traefik.yml.template:19,174 | agents-api-chain、api-jwt-auth |
| 服务拓扑组装 | docker-compose.yml:6 | include |