# Python 3.12+, uv-managed venv
uv venv --python 3.12
uv pip install -e ".[dev]"Package manager is uv, not pip. The uv.lock file is checked in and authoritative for dependency versions.
# All tests (mock-based, no API key needed) — 62 tests
.venv/bin/python -m pytest tests/ -v
# CLI (interactive)
.venv/bin/python -m agentframe
# or after `uv pip install -e .`:
afcliagentframe/
core/agent.py # Agent class + 4 hooks + LangGraph StateGraph
llm/client.py # openai wrapper: invoke, ainvoke, stream, astream
tools/ # function_tool decorator, ToolRegistry, MCP client
compression/ # Token-threshold summarization (Compressor: compress + acompress)
memory/hooks.py # doc-only: users pass langgraph BaseCheckpointSaver
cli/ # ChatAgent, ~/.afcli.toml config, UTF-8 input
tests/ # 62 tests across 7 files
The CLI does NOT use the LangGraph StateGraph. It calls agent.ainvoke() which routes through _acall_agent → llm_client.astream → hooks fire per token. The graph is for library users calling invoke()/ainvoke()/stream()/astream().
Sync path: invoke() → _build_graph() → _call_agent (uses llm_client.invoke) → _should_continue → _call_tools → loop
Async path: ainvoke() → _abuild_graph() → _acall_agent (uses llm_client.astream + hooks) → _ashould_continue → _acall_tools → loop
Shared helpers reduce sync/async duplication:
_build_graph_impl(agent_node, tools_node, should_continue_fn)— graphs for both paths_prepare_agent_state(state)/_prepare_agent_state_async(state)— compression + tool list_process_tool_calls(messages, last_message)— approved_ids + rejection messages
The async path MUST extract tool_calls from the "done" stream event. Both sync and async _call_tools MUST call on_tool_result. Do NOT remove either — tests catch this.
| Hook | Signature | Fires when |
|---|---|---|
on_llm_reasoning |
(text: str) |
each reasoning chunk during streaming |
on_llm_content |
(text: str) |
each content chunk during streaming |
on_tool_call |
(tool_calls: list[dict]) -> list[dict] |
before tool execution, return approved subset |
on_tool_result |
(name: str, result: str) |
after each tool execution |
Default implementations are no-ops. All four fire in both sync and async paths.
- Agent tests:
patch.object(agent.llm_client, "invoke", ...)orpatch.object(agent.llm_client, "astream", ...). Do NOT patchopenaimodule-level — always patch on the instance. - LLMClient tests:
patch.object(client, "_get_client")/patch.object(client, "_get_aclient")to mock the inneropenaiclient.
astream returns an async generator. Mock with a closure that yields events per call:
def _make_astream(*event_lists):
calls = iter(event_lists)
async def mock(messages, tools=None):
for event in next(calls):
yield event
return mockConftest provides make_response() and make_tool_call() helpers. Both async and sync paths have test coverage (test_agent.py + test_agent_async.py).
代码修改完成后,运行基于 pyright 检查:
.venv/bin/basedpyright agentframe/- 可修复的告警/报错直接修复
- 修复代价大或无法修复的,在行末加
# type: ignore[<error-code>]屏蔽(如# type: ignore[arg-type]、# type: ignore[reportMissingImports]) - 防御性代码(如类型标注保证安全但仍保留的运行时检查)导致的告警也可用
# type: ignore[unreachable]屏蔽 - 可选依赖导致的导入错误使用
# type: ignore[reportMissingImports]屏蔽
- LSP import errors: LSP runs outside the venv, so
langchain_core/openai/langgraphimports show as unresolved. Ignore them. base_urlsupport:LLMClientacceptsbase_urlfor custom endpoints (Ollama, vLLM, etc.). Pass it toAgentand it flows through toopenai.OpenAI(base_url=...).- basedpyright false positive:
_build_graph_implparameter types (Callable[[AgentState], dict]) cause a false-positive error onworkflow.add_node("agent", agent_node)because LangGraph'sStateNodeProtocol expectsstateas a keyword param. Code runs correctly; this is a LangGraph type-stub limitation. - Config file:
~/.afcli.tomlauto-created on first CLI run.api_keydefaults to""— set it or useLLM_AUTH_KEY/OPENAI_API_KEYenv var. - Session persistence: pass
session_id="name"toinvoke()/ainvoke(). Must also pass acheckpointer(e.g.MemorySaverorSqliteSaver) when constructing Agent. - CLI builtin tools: The CLI auto-registers
bashtool (agentframe/tools/builtin/bash.py). - MCP connection caching:
_ensure_mcp_connected()lazily connects and reuses MCP clients. Callagent.aclose_mcp()to clean up subprocesses. - Dict tools in ToolRegistry: registering a
dicttool withoutfunction.nameraisesValueError(no silent fallback). - Compressor dual path:
Compressor.compress()is sync,Compressor.acompress()is async. The async agent path usesacompressto avoid blocking the event loop.
In repositories indexed by CodeGraph (a .codegraph/ directory exists at the repo root), reach for it BEFORE grep/find or reading files when you need to understand or locate code:
- MCP tool (when available):
codegraph_exploreanswers most code questions in one call — the relevant symbols' verbatim source plus the call paths between them, including dynamic-dispatch hops grep can't follow. Name a file or symbol in the query to read its current line-numbered source. If it's listed but deferred, load it by name via tool search. - Shell (always works):
codegraph explore "<symbol names or question>"prints the same output.
If there is no .codegraph/ directory, skip CodeGraph entirely — indexing is the user's decision.