LightAgent: Lightweight Python framework for OpenAI-compatible agents with tools, memory, guardrails, tracing, lifecycle hooks, multi-agent collaboration, and workflows.
1,217
stars
153
commits
Python
primary language
Sep 4, 2026
updated

LightAgent is an ultra‑lightweight, open‑source framework that now natively supports Skills — letting you compose reusable capabilities with persistent memory, tool use, and tree‑of‑thought reasoning. It streamlines multi‑agent collaboration (build self‑learning agents in one step), connects to MCP over stdio and SSE, runs on any modern LLM (OpenAI, DeepSeek, Qwen, and more), and outputs OpenAI‑compatible streaming APIs for instant drop‑in with any chat interface. Small, modular, and skill‑ready — spin it up in five minutes.
max_tool_iterations, consistent on_error / after_run closure, and expanded regression coverage.agent.export_trace(), and prompt-safe model request summaries for production debugging.agent.run() and stream=True behavior compatible.Older release notes are available on GitHub Releases.

mem0 memory module, automatically managing user personalized memory during conversations, making agents smarter.Tools) and MCP tool integration, flexible expansion to meet diverse needs.ToT) module with reflection, supporting complex task decomposition and multi-step reasoning, enhancing task processing capabilities.trace=True run traces record structured run lifecycle, model request summaries, tool calls, tool results, and errors without changing the default string return value.LightEvaluator runs deterministic agent or LightFlow regression cases for output, tool choice, policy events, recovery, latency, usage, and estimated cost.HumanApprovalHook, durable review stores, and LightFlow approval checkpoints support approve, reject, argument editing, human responses, batches, and trace feedback for high-impact actions.hooks=[...] middleware can observe, replace, or block run, model, tool, memory, and LightFlow step phases while recording hook decisions in trace events.| Layer | Main API | Use it when you need |
|---|---|---|
| Single agent runtime | LightAgent | One agent with model calls, tools, memory, streaming, trace, and guardrails. |
| Durable runtime state | Session, AgentRuntime | Replayable events, checkpoints, Inbox, Goals, Budgets, Jobs, and context recovery. |
| Capability layer | CapabilityRegistry, PolicyEngine | Scoped Providers, permission snapshots, lifecycle, policy, and audit. |
| Multi-agent routing | LightSwarm | Role-based delegation across specialized agents. |
| Deterministic workflow | LightFlow | Ordered DAG workflows, retries, checkpoints, durable approvals, resume, and rerun. |
| Tools and integrations | tools, ToolRegistry, MCP | Python tools, generated tools, runtime tool loading, or MCP tool servers. |
| Memory boundary | MemoryPolicy, MemoryScope | Tenant isolation, provenance, trust, expiration, and write admission controls. |
| Shared memory prototype | SharedMemoryPool | In-memory shared memory experiments across agents. |
| Safety controls | input_guardrails, tool_guardrails, output_guardrails | Privacy blocking, sensitive tool confirmation, high-risk parameter checks, and output redaction. |
| Runtime hooks | hooks, HookContext, HookDecision | Policy, audit, redaction, routing, and payload mutation at lifecycle boundaries. |
| Observability | trace=True, summarize_trace, trace exporters | Structured events, latency, usage, retry/error metrics, JSONL or external export. |
| Evaluation | LightEvaluator, EvaluationCase | Fixed regression cases for output, tools, policy events, recovery, latency, and cost. |
| Human review | HumanApprovalHook, ApprovalDecision, review stores | Fail-closed tool/handoff review, argument editing, durable decisions, and feedback. |
LightAgent keeps the default call path simple while allowing production controls to be added incrementally.
| Pattern | Minimal call | Notes |
|---|---|---|
| Basic response | agent.run(query) | Returns a string by default. |
| Streaming | agent.run(query, stream=True) | Returns OpenAI-compatible streaming chunks. |
| Structured result | agent.run(query, result_format="object") | Returns content plus structured metadata. |
| Trace | agent.run(query, trace=True) | Records events without changing the default string return. |
| User memory | agent.run(query, user_id="alice") | Uses the configured memory backend and memory policy. |
| Tools | LightAgent(..., tools=[fn]) | Functions should expose tool_info metadata. |
| Guardrails | LightAgent(..., input_guardrails=[...]) | Add input, tool, and output policies per agent. |
| Runtime hooks | LightAgent(..., hooks=[fn]) | Observe, replace, or block lifecycle payloads. |
| Evaluation | LightEvaluator().run(agent, cases) | Run deterministic behavioral checks from structured traces. |
| Tool approval | LightAgent(..., hooks=[HumanApprovalHook(...)]) | Require review before selected tools or handoffs. |
| Workflow | LightFlow().step(...).run(query) | Use for deterministic multi-step execution. |
| Durable session | agent.run(query, session_id="project-42") | Continue and replay a persisted conversation. |
| Async | await agent.arun(query) | Keep an asyncio application responsive. |
from LightAgent import EvaluationCase, HumanApprovalHook, LightEvaluator
agent.hooks.hooks.append(HumanApprovalHook(tools={"send_payment"}))
report = LightEvaluator().run(agent, [
EvaluationCase(
name="payment policy",
query="Pay invoice 42",
expected_trace_events=("approval_required",),
forbidden_tools=("delete_account",),
),
])
print(report.to_dict())
Pending approvals can be stored in JsonReviewStore; resolve them with an
ApprovalDecision, then rerun with approval_id=request.request_id. LightFlow
also persists step approvals through flow.approve(...); flow.resume(...).
If you are using LightSwarm or other multi-agent patterns and start seeing role drift, cross-agent memory issues or confusing logs, you can check the
Multi-agent failure map for a small symptom → mode → debug checklist.
This page is docs-only and does not change any framework code.
For common installation, model provider, tool, memory, MCP, Skills, streaming, and LightSwarm questions, see FAQ.
For deterministic multi-step workflows, checkpointed run records, resume/rerun, approval nodes, fallback agents, and step status tracking, see LightFlow.
For custom tool creation, runtime tools, ToolRegistry, ToolLoader, AsyncToolDispatcher, and MCP tool integration, see Tools Guide.
For packaging existing Tools, Skills, MCP settings, Hooks, and memory adapters into an offline-validatable extension, see Lightweight Connectors.
For shared long-term memory or graph memory deployments, review the Memory Security Guidance.
For the opt-in Mem0 Graph backend security matrix and issue #39 validation boundary, see Shared Graph Memory Security Validation.
For lightweight shared memory experiments, see SharedMemoryPool.
For optional ClawMem long-term memory adapter setup, see ClawMem Memory Adapter.
For memory write admission, expiration-aware retrieval, and low-quality memory write blocking, see Memory Admission And Mutation Controls.
For separating trace, user memory, self-reflection memory, and LightSwarm delegation state, see Memory, Trace, And Swarm Boundaries.
For input, tool, and output safety policies, see Guardrails.
For runtime middleware that can observe, replace, or block lifecycle payloads, see Runtime Hooks.
For OpenRouter, local LLM, and OpenAI-compatible provider setup, see Model Provider Configuration.
For structured error codes and troubleshooting hints, see Error Handling.
For trace observability, summaries, and external export, see Trace Observability.
For deterministic regression cases, metrics, and CI guidance, see Evaluation Harness.
For tool/handoff approval, durable LightFlow review, batches, and feedback, see Human Review.
For the v1.0 stability proposal, supported Python versions, public imports, and compatibility promises, see Public API And Compatibility Inventory.
For durable Sessions, Capability Providers, Policy, Inbox, Goals, Budgets, Jobs, compaction, subagents, Skills/MCP updates, and SQLite FTS5 retrieval, see LightAgent v0.10 Runtime.
For browser-use integration with recent browser-use versions, see browser-use Integration.
pip install lightagent
(Optional installation) Install the Mem0 package via pip:
pip install mem0ai
Alternatively, you can use Mem0 on a hosted platform by clicking here.
from LightAgent import LightAgent
# Initialize Agent
agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
# Run Agent
response = agent.run("Hello, who are you?")
print(response)
Tracing is opt-in and keeps the default agent.run() behavior backward compatible.
from LightAgent import LightAgent
agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
result = agent.run("Hello, who are you?", result_format="object", trace=True)
print(result.content)
print(result.trace_id)
print(result.trace)
for event in agent.export_trace():
print(event["type"], event["data"])
LightFlow can persist workflow checkpoints and resume failed runs without
starting from the first step again.
from LightAgent import JsonLightFlowStore, LightAgent, LightFlow
research_agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
writer_agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
store = JsonLightFlowStore(".lightflow_runs")
flow = (
LightFlow(store=store)
.step("research", agent=research_agent, timeout=30)
.step("write", agent=writer_agent, depends_on=["research"], max_retry=2)
)
result = flow.run("Analyze this company", run_id="report-001", trace=True)
if not result.success:
result = flow.resume("report-001")
print(result.status)
print(flow.get_run("report-001")["steps"])
SharedMemoryPool is a lightweight in-memory prototype for multi-agent shared
memory experiments.
from LightAgent import LightAgent, MemoryPolicy, SharedMemoryPool
shared_memory = SharedMemoryPool(agent_name="writer")
agent = LightAgent(
name="writer",
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
memory=shared_memory,
memory_policy=MemoryPolicy(
namespace="tenant-a",
allow_unattributed_results=False,
allowed_sources=("user",),
allowed_scopes=("user",),
),
)
agent.run("Remember that I prefer concise reports.", user_id="alice")
print(shared_memory.list_records(user_id="tenant-a:alice"))
from LightAgent import LightAgent
# Initialize Agent
agent = LightAgent(
role="Please remember that you are LightAgent, a useful assistant that helps users use multiple tools.", # system role description
model="gpt-4.1", # Supported models: openai, chatglm, deepseek, qwen, etc.
api_key="your_api_key", # Replace with your large model provider API Key
base_url="your_base_url", # Replace with your large model provider api url
)
# Run Agent
response = agent.run("Who are you?")
print(response)
from LightAgent import LightAgent
# Define Tool
def get_weather(city_name: str) -> str:
"""
Get the current weather for `city_name`
"""
return f"Query result: {city_name} is sunny."
# Define tool information inside the function
get_weather.tool_info = {
"tool_name": "get_weather",
"tool_description": "Get current weather information for the specified city.",
"tool_params": [
{"name": "city_name", "description": "The name of the city to query", "type": "string", "required": True},
]
}
tools = [get_weather]
# Initialize Agent
agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url", tools=tools)
# Run Agent
response = agent.run("Please check the weather in Shanghai.")
print(response)
Supports an unlimited number of customizable tools.
Multiple tool examples: tools = [search_news, get_weather, get_stock_realtime_data, get_stock_kline_data]
README keeps the core usage model in one place. Longer examples, adapter-specific setup, and production guidance live in the dedicated docs pages.
mem0)LightAgent accepts any memory backend that provides store(data, user_id) and retrieve(query, user_id). This keeps memory detachable: you can start with a simple custom class, use mem0, or plug in a vector/graph memory adapter without changing agent code.
Use user_id to isolate conversations, and use MemoryPolicy when memory is shared across users, tenants, agents, or traces.
from LightAgent import LightAgent, MemoryPolicy
agent = LightAgent(
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
memory=your_memory_backend,
memory_policy=MemoryPolicy(
namespace="tenant-a",
allowed_sources=("user", "reflection"),
allowed_scopes=("user", "agent"),
reject_duplicate_writes=True,
min_write_length=8,
),
)
response = agent.run("Remember that I prefer concise reports.", user_id="alice")
See Memory Security Guidance, Memory Admission And Mutation Controls, and Memory, Trace, And Swarm Boundaries.
Use Python functions with tool_info metadata to expose controlled capabilities to an agent. LightAgent can also work with runtime tools, ToolRegistry, ToolLoader, AsyncToolDispatcher, generated tools, and MCP tools.
from LightAgent import LightAgent
def get_order_status(order_id: str) -> str:
return f"Order {order_id} is being processed."
get_order_status.tool_info = {
"tool_name": "get_order_status",
"tool_description": "Get the status of an order.",
"tool_params": [
{"name": "order_id", "description": "Order ID", "type": "string", "required": True},
],
}
agent = LightAgent(
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
tools=[get_order_status],
)
For advanced tool loading, generated tools, async dispatch, and MCP integration, see Tools Guide.
agent.create_tool() can generate tool code from API documentation or natural-language descriptions. It is useful when converting an internal API document into callable Python tools.
Keep generated tools in a reviewed tools directory, test them before production use, and avoid committing generated or experimental local tools unless they are part of the public package.
agent.create_tool(
"Create a tool that calls the internal order status API.",
tools_directory="tools",
)
Enable tree_of_thought=True when a task needs explicit planning and reflection before tool use or final response generation. ToT is best for complex multi-step reasoning, tool selection, and tasks where the agent should inspect its plan before acting.
agent = LightAgent(
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
tree_of_thought=True,
tot_model="gpt-4.1",
tot_api_key="your_api_key",
tot_base_url="your_base_url",
)
LightSwarm routes work across specialized agents for role-based delegation and multi-agent task handling. Use it when one front-facing agent should delegate to domain agents such as support, HR, finance, research, or data analysis.
Keep each agent's role narrow, give each agent clear boundaries, and avoid letting delegated agents write unrelated long-term memory unless the memory policy allows it.
from LightAgent import LightAgent, LightSwarm
frontdesk = LightAgent(name="frontdesk", role="Route requests to the right specialist.")
finance = LightAgent(name="finance", role="Answer finance approval questions.")
swarm = LightSwarm()
swarm.register_agent(frontdesk, finance)
result = swarm.run(agent=frontdesk, query="How do I approve a vendor payment?")
For debugging role drift or cross-agent memory issues, see Multi-agent failure map.
agent.run(query, stream=True) returns OpenAI-compatible streaming chunks and remains backward compatible with existing integrations. Use it for chat UIs, long responses, and agent services that should start sending output before the full answer is complete.
For tool-heavy streaming tasks, max_tool_iterations can cap repeated tool-call loops independently from max_retry.
for chunk in agent.run("Write a short report about AI agents.", stream=True):
print(chunk)
Self-learning can be combined with a memory backend so an agent can retain scenario-specific knowledge across users or sessions. In production, pair self-learning with MemoryPolicy so low-quality, private, expired, or unrelated content does not enter long-term memory.
Recommended production controls:
namespace to separate tenants or environments.allowed_sources, allowed_scopes, and allowed_agent_names to limit retrieved memory.memory_write_admission, reject_write_patterns, min_write_length, and reject_duplicate_writes to control writes.enforce_expires_at=True when memory records carry expiration metadata.LightAgent can send trace data to Langfuse through tracetools configuration. For most local debugging, start with built-in trace events:
result = agent.run("Debug this workflow.", result_format="object", trace=True)
print(result.trace_id)
print(agent.export_trace())
See Trace Observability for event shapes and production debugging guidance.
LightEvaluator measures output, tool selection, policy events, recovery, latency, usage, and estimated cost against deterministic business cases.
LightFlow is the deterministic workflow layer. Use it when a task should run through known steps instead of relying on free-form agent planning.
Key v0.9.0 workflow controls:
pending, running, success, failed, skipped, waiting_approval.flow.validate(strict=True) checks unknown dependencies, cycles, and isolated steps.timeout, max_retry, cancel_if, fallback_agent, requires_approval, and approval_handler.JsonLightFlowStore stores run records and checkpoints.flow.resume(run_id) continues incomplete runs; flow.rerun_step(run_id, step_name) reruns one step and downstream steps.flow.get_run(run_id) and flow.list_runs() expose workflow state for UIs.from LightAgent import JsonLightFlowStore, LightAgent, LightFlow
store = JsonLightFlowStore(".lightflow_runs")
draft_agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
review_agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
backup_agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
def approve(step, context):
return {"approved": True, "reason": "Approved by policy."}
flow = (
LightFlow(store=store)
.step("draft", agent=draft_agent, timeout=30, max_retry=2)
.step(
"review",
agent=review_agent,
depends_on=["draft"],
requires_approval=True,
approval_handler=approve,
fallback_agent=backup_agent,
)
)
validation = flow.validate(strict=True)
if validation["errors"]:
raise ValueError(validation["errors"])
result = flow.run("Prepare a customer support summary.", run_id="case-001", trace=True)
record = flow.get_run("case-001")
See LightFlow for complete workflow examples.
Guardrails are lightweight policy hooks around agent execution:
from LightAgent import (
LightAgent,
high_risk_parameter_guardrail,
output_redaction_guardrail,
privacy_input_guardrail,
sensitive_tool_confirmation_guardrail,
)
agent = LightAgent(
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
input_guardrails=[privacy_input_guardrail()],
tool_guardrails=[
sensitive_tool_confirmation_guardrail(["transfer_money"], approved=False),
high_risk_parameter_guardrail({
"amount": lambda value: value is not None and float(value) <= 1000,
}),
],
output_guardrails=[output_redaction_guardrail()],
)
Use default guardrail templates for privacy-sensitive input, sensitive tool confirmation, high-risk parameter validation, and output redaction. See Guardrails.
Runtime hooks are ordered middleware for production policies that need to observe, replace, or block lifecycle payloads without changing the default agent.run() behavior.
from LightAgent import HookDecision, LightAgent, PolicyHook
def redact_before_model(ctx):
if ctx.phase != "before_model_request":
return None
params = dict(ctx.payload["params"])
messages = list(params["messages"])
messages[-1] = {
**messages[-1],
"content": messages[-1]["content"].replace("secret-token", "[REDACTED]"),
}
params["messages"] = messages
return HookDecision.replace({"params": params})
def block_dangerous_tool(ctx):
if ctx.phase == "before_tool_call" and ctx.payload["tool_name"] == "delete_file":
return HookDecision.block("delete_file requires manual approval")
return None
agent = LightAgent(
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
hooks=[
redact_before_model,
PolicyHook(block_dangerous_tool, phases={"before_tool_call"}, timeout=2.0),
],
)
result = agent.run("Summarize this secret-token safely.", result_format="object", trace=True)
print(result.content)
print(result.trace)
Hooks can target before_run, after_run, on_error, before_model_request, after_model_response, before_tool_call, after_tool_result, before_memory_retrieve, after_memory_retrieve, before_memory_write, after_memory_write, before_memory_promote, after_memory_promote, and on_handoff. Wrap security-sensitive callbacks with PolicyHook(...) to fail closed on exceptions or timeouts while ordinary observability hooks remain failure-isolated. LightFlow(hooks=[...]) also supports step lifecycle hooks such as before_flow_step, after_flow_step, on_approval_required, on_resume, and on_rerun. See Runtime Hooks and Runtime Hook Recipes.
SharedMemoryPool is an in-memory shared memory prototype for multi-agent experiments. It is append-first and keeps provenance metadata, making it useful for testing how multiple agents share information before adopting a durable vector or graph memory backend.
Use it with MemoryPolicy so each agent retrieves only memory that matches the expected namespace, source, scope, trust, confidence, or agent name.
Self-learning and delegation evidence become non-injectable memory candidates first; promote them with MemoryPolicy(memory_promotion_admission=...) or agent.promote_memory_candidate(candidate_id) before they can enter future prompts.
When upgrading from v0.9.4, audit legacy internal memory before backfilling promotion_status="promoted" and injectable=True; use require_promotion_for_internal_memory=False only as a temporary compatibility option. See Memory Admission.
LightAgent works with OpenAI-compatible chat completion endpoints, including OpenAI, OpenRouter, Zhipu ChatGLM, DeepSeek, Qwen, StepFun, Moonshot/Kimi, MiniMax, vLLM, llama.cpp, Ollama-compatible endpoints, and self-hosted gateways.
For provider-specific parameters, base URLs, local model setup, and troubleshooting, see Model Provider Configuration.
We welcome any form of contribution! Whether it's code, documentation, tests, or feedback, it's a tremendous help to the project. If you have great ideas or find bugs, please submit an Issue or Pull Request. Here are the contribution steps:
Fork button at the top right corner to copy the project to your GitHub repository.git checkout -b feature/YourFeature
git commit -m 'Add some feature'
git push origin feature/YourFeature
We will review your contributions promptly. Thank you for your support! ❤️
Shanghai Wanxing AI and Professor Zhang Liwen's research group from the School of Statistics and Data Science at Shanghai University of Finance and Economics have jointly open-sourced a new generation intelligent agent framework called LightAgent.The development and implementation of LightAgent owe much to the inspiration and support from the following open-source projects, especially the outstanding projects and teams:
LightAgent is licensed under the Apache 2.0 License. You can freely use, modify, and distribute this project, but please adhere to the terms of the license.
If you have any questions or suggestions, please feel free to contact Wanxing AI or Professor Zhang Liwen from the School of Statistics and Data Science at Shanghai University of Finance and Economics:
We look forward to your feedback and work together to make LightAgent even stronger! 🚀
LightAgent - Making intelligence lighter, making the future simpler. 🌈
LightAgent —— A lightweight, flexible, and powerful active Agent framework that assists you in quickly building intelligent applications!
@misc{2509.09292,
Author = {Weige Cai and Tong Zhu and Jinyi Niu and Ruiqi Hu and Lingyao Li and Tenglong Wang and Xiaowu Dai and Weining Shen and Liwen Zhang},
Title = {LightAgent: Production-level Open-source Agentic AI Framework},
Year = {2025},
Eprint = {arXiv:2509.09292},
Eprinttype = {arXiv},
Eprintclass = {cs.AI},
Url = {https://arxiv.org/abs/2509.09292},
Doi = {10.48550/arXiv.2509.09292},
Note = {Submitted on 11 Sep 2025}
}
Python
99.9%
LightAgent: Lightweight Python framework for OpenAI-compatible agents with tools, memory, guardrails, tracing, lifecycle hooks, multi-agent collaboration, and workflows.
1,217
stars
153
commits
Python
primary language
Sep 4, 2026
updated

LightAgent is an ultra‑lightweight, open‑source framework that now natively supports Skills — letting you compose reusable capabilities with persistent memory, tool use, and tree‑of‑thought reasoning. It streamlines multi‑agent collaboration (build self‑learning agents in one step), connects to MCP over stdio and SSE, runs on any modern LLM (OpenAI, DeepSeek, Qwen, and more), and outputs OpenAI‑compatible streaming APIs for instant drop‑in with any chat interface. Small, modular, and skill‑ready — spin it up in five minutes.
max_tool_iterations, consistent on_error / after_run closure, and expanded regression coverage.agent.export_trace(), and prompt-safe model request summaries for production debugging.agent.run() and stream=True behavior compatible.Older release notes are available on GitHub Releases.

mem0 memory module, automatically managing user personalized memory during conversations, making agents smarter.Tools) and MCP tool integration, flexible expansion to meet diverse needs.ToT) module with reflection, supporting complex task decomposition and multi-step reasoning, enhancing task processing capabilities.trace=True run traces record structured run lifecycle, model request summaries, tool calls, tool results, and errors without changing the default string return value.LightEvaluator runs deterministic agent or LightFlow regression cases for output, tool choice, policy events, recovery, latency, usage, and estimated cost.HumanApprovalHook, durable review stores, and LightFlow approval checkpoints support approve, reject, argument editing, human responses, batches, and trace feedback for high-impact actions.hooks=[...] middleware can observe, replace, or block run, model, tool, memory, and LightFlow step phases while recording hook decisions in trace events.| Layer | Main API | Use it when you need |
|---|---|---|
| Single agent runtime | LightAgent | One agent with model calls, tools, memory, streaming, trace, and guardrails. |
| Durable runtime state | Session, AgentRuntime | Replayable events, checkpoints, Inbox, Goals, Budgets, Jobs, and context recovery. |
| Capability layer | CapabilityRegistry, PolicyEngine | Scoped Providers, permission snapshots, lifecycle, policy, and audit. |
| Multi-agent routing | LightSwarm | Role-based delegation across specialized agents. |
| Deterministic workflow | LightFlow | Ordered DAG workflows, retries, checkpoints, durable approvals, resume, and rerun. |
| Tools and integrations | tools, ToolRegistry, MCP | Python tools, generated tools, runtime tool loading, or MCP tool servers. |
| Memory boundary | MemoryPolicy, MemoryScope | Tenant isolation, provenance, trust, expiration, and write admission controls. |
| Shared memory prototype | SharedMemoryPool | In-memory shared memory experiments across agents. |
| Safety controls | input_guardrails, tool_guardrails, output_guardrails | Privacy blocking, sensitive tool confirmation, high-risk parameter checks, and output redaction. |
| Runtime hooks | hooks, HookContext, HookDecision | Policy, audit, redaction, routing, and payload mutation at lifecycle boundaries. |
| Observability | trace=True, summarize_trace, trace exporters | Structured events, latency, usage, retry/error metrics, JSONL or external export. |
| Evaluation | LightEvaluator, EvaluationCase | Fixed regression cases for output, tools, policy events, recovery, latency, and cost. |
| Human review | HumanApprovalHook, ApprovalDecision, review stores | Fail-closed tool/handoff review, argument editing, durable decisions, and feedback. |
LightAgent keeps the default call path simple while allowing production controls to be added incrementally.
| Pattern | Minimal call | Notes |
|---|---|---|
| Basic response | agent.run(query) | Returns a string by default. |
| Streaming | agent.run(query, stream=True) | Returns OpenAI-compatible streaming chunks. |
| Structured result | agent.run(query, result_format="object") | Returns content plus structured metadata. |
| Trace | agent.run(query, trace=True) | Records events without changing the default string return. |
| User memory | agent.run(query, user_id="alice") | Uses the configured memory backend and memory policy. |
| Tools | LightAgent(..., tools=[fn]) | Functions should expose tool_info metadata. |
| Guardrails | LightAgent(..., input_guardrails=[...]) | Add input, tool, and output policies per agent. |
| Runtime hooks | LightAgent(..., hooks=[fn]) | Observe, replace, or block lifecycle payloads. |
| Evaluation | LightEvaluator().run(agent, cases) | Run deterministic behavioral checks from structured traces. |
| Tool approval | LightAgent(..., hooks=[HumanApprovalHook(...)]) | Require review before selected tools or handoffs. |
| Workflow | LightFlow().step(...).run(query) | Use for deterministic multi-step execution. |
| Durable session | agent.run(query, session_id="project-42") | Continue and replay a persisted conversation. |
| Async | await agent.arun(query) | Keep an asyncio application responsive. |
from LightAgent import EvaluationCase, HumanApprovalHook, LightEvaluator
agent.hooks.hooks.append(HumanApprovalHook(tools={"send_payment"}))
report = LightEvaluator().run(agent, [
EvaluationCase(
name="payment policy",
query="Pay invoice 42",
expected_trace_events=("approval_required",),
forbidden_tools=("delete_account",),
),
])
print(report.to_dict())
Pending approvals can be stored in JsonReviewStore; resolve them with an
ApprovalDecision, then rerun with approval_id=request.request_id. LightFlow
also persists step approvals through flow.approve(...); flow.resume(...).
If you are using LightSwarm or other multi-agent patterns and start seeing role drift, cross-agent memory issues or confusing logs, you can check the
Multi-agent failure map for a small symptom → mode → debug checklist.
This page is docs-only and does not change any framework code.
For common installation, model provider, tool, memory, MCP, Skills, streaming, and LightSwarm questions, see FAQ.
For deterministic multi-step workflows, checkpointed run records, resume/rerun, approval nodes, fallback agents, and step status tracking, see LightFlow.
For custom tool creation, runtime tools, ToolRegistry, ToolLoader, AsyncToolDispatcher, and MCP tool integration, see Tools Guide.
For packaging existing Tools, Skills, MCP settings, Hooks, and memory adapters into an offline-validatable extension, see Lightweight Connectors.
For shared long-term memory or graph memory deployments, review the Memory Security Guidance.
For the opt-in Mem0 Graph backend security matrix and issue #39 validation boundary, see Shared Graph Memory Security Validation.
For lightweight shared memory experiments, see SharedMemoryPool.
For optional ClawMem long-term memory adapter setup, see ClawMem Memory Adapter.
For memory write admission, expiration-aware retrieval, and low-quality memory write blocking, see Memory Admission And Mutation Controls.
For separating trace, user memory, self-reflection memory, and LightSwarm delegation state, see Memory, Trace, And Swarm Boundaries.
For input, tool, and output safety policies, see Guardrails.
For runtime middleware that can observe, replace, or block lifecycle payloads, see Runtime Hooks.
For OpenRouter, local LLM, and OpenAI-compatible provider setup, see Model Provider Configuration.
For structured error codes and troubleshooting hints, see Error Handling.
For trace observability, summaries, and external export, see Trace Observability.
For deterministic regression cases, metrics, and CI guidance, see Evaluation Harness.
For tool/handoff approval, durable LightFlow review, batches, and feedback, see Human Review.
For the v1.0 stability proposal, supported Python versions, public imports, and compatibility promises, see Public API And Compatibility Inventory.
For durable Sessions, Capability Providers, Policy, Inbox, Goals, Budgets, Jobs, compaction, subagents, Skills/MCP updates, and SQLite FTS5 retrieval, see LightAgent v0.10 Runtime.
For browser-use integration with recent browser-use versions, see browser-use Integration.
pip install lightagent
(Optional installation) Install the Mem0 package via pip:
pip install mem0ai
Alternatively, you can use Mem0 on a hosted platform by clicking here.
from LightAgent import LightAgent
# Initialize Agent
agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
# Run Agent
response = agent.run("Hello, who are you?")
print(response)
Tracing is opt-in and keeps the default agent.run() behavior backward compatible.
from LightAgent import LightAgent
agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
result = agent.run("Hello, who are you?", result_format="object", trace=True)
print(result.content)
print(result.trace_id)
print(result.trace)
for event in agent.export_trace():
print(event["type"], event["data"])
LightFlow can persist workflow checkpoints and resume failed runs without
starting from the first step again.
from LightAgent import JsonLightFlowStore, LightAgent, LightFlow
research_agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
writer_agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
store = JsonLightFlowStore(".lightflow_runs")
flow = (
LightFlow(store=store)
.step("research", agent=research_agent, timeout=30)
.step("write", agent=writer_agent, depends_on=["research"], max_retry=2)
)
result = flow.run("Analyze this company", run_id="report-001", trace=True)
if not result.success:
result = flow.resume("report-001")
print(result.status)
print(flow.get_run("report-001")["steps"])
SharedMemoryPool is a lightweight in-memory prototype for multi-agent shared
memory experiments.
from LightAgent import LightAgent, MemoryPolicy, SharedMemoryPool
shared_memory = SharedMemoryPool(agent_name="writer")
agent = LightAgent(
name="writer",
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
memory=shared_memory,
memory_policy=MemoryPolicy(
namespace="tenant-a",
allow_unattributed_results=False,
allowed_sources=("user",),
allowed_scopes=("user",),
),
)
agent.run("Remember that I prefer concise reports.", user_id="alice")
print(shared_memory.list_records(user_id="tenant-a:alice"))
from LightAgent import LightAgent
# Initialize Agent
agent = LightAgent(
role="Please remember that you are LightAgent, a useful assistant that helps users use multiple tools.", # system role description
model="gpt-4.1", # Supported models: openai, chatglm, deepseek, qwen, etc.
api_key="your_api_key", # Replace with your large model provider API Key
base_url="your_base_url", # Replace with your large model provider api url
)
# Run Agent
response = agent.run("Who are you?")
print(response)
from LightAgent import LightAgent
# Define Tool
def get_weather(city_name: str) -> str:
"""
Get the current weather for `city_name`
"""
return f"Query result: {city_name} is sunny."
# Define tool information inside the function
get_weather.tool_info = {
"tool_name": "get_weather",
"tool_description": "Get current weather information for the specified city.",
"tool_params": [
{"name": "city_name", "description": "The name of the city to query", "type": "string", "required": True},
]
}
tools = [get_weather]
# Initialize Agent
agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url", tools=tools)
# Run Agent
response = agent.run("Please check the weather in Shanghai.")
print(response)
Supports an unlimited number of customizable tools.
Multiple tool examples: tools = [search_news, get_weather, get_stock_realtime_data, get_stock_kline_data]
README keeps the core usage model in one place. Longer examples, adapter-specific setup, and production guidance live in the dedicated docs pages.
mem0)LightAgent accepts any memory backend that provides store(data, user_id) and retrieve(query, user_id). This keeps memory detachable: you can start with a simple custom class, use mem0, or plug in a vector/graph memory adapter without changing agent code.
Use user_id to isolate conversations, and use MemoryPolicy when memory is shared across users, tenants, agents, or traces.
from LightAgent import LightAgent, MemoryPolicy
agent = LightAgent(
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
memory=your_memory_backend,
memory_policy=MemoryPolicy(
namespace="tenant-a",
allowed_sources=("user", "reflection"),
allowed_scopes=("user", "agent"),
reject_duplicate_writes=True,
min_write_length=8,
),
)
response = agent.run("Remember that I prefer concise reports.", user_id="alice")
See Memory Security Guidance, Memory Admission And Mutation Controls, and Memory, Trace, And Swarm Boundaries.
Use Python functions with tool_info metadata to expose controlled capabilities to an agent. LightAgent can also work with runtime tools, ToolRegistry, ToolLoader, AsyncToolDispatcher, generated tools, and MCP tools.
from LightAgent import LightAgent
def get_order_status(order_id: str) -> str:
return f"Order {order_id} is being processed."
get_order_status.tool_info = {
"tool_name": "get_order_status",
"tool_description": "Get the status of an order.",
"tool_params": [
{"name": "order_id", "description": "Order ID", "type": "string", "required": True},
],
}
agent = LightAgent(
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
tools=[get_order_status],
)
For advanced tool loading, generated tools, async dispatch, and MCP integration, see Tools Guide.
agent.create_tool() can generate tool code from API documentation or natural-language descriptions. It is useful when converting an internal API document into callable Python tools.
Keep generated tools in a reviewed tools directory, test them before production use, and avoid committing generated or experimental local tools unless they are part of the public package.
agent.create_tool(
"Create a tool that calls the internal order status API.",
tools_directory="tools",
)
Enable tree_of_thought=True when a task needs explicit planning and reflection before tool use or final response generation. ToT is best for complex multi-step reasoning, tool selection, and tasks where the agent should inspect its plan before acting.
agent = LightAgent(
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
tree_of_thought=True,
tot_model="gpt-4.1",
tot_api_key="your_api_key",
tot_base_url="your_base_url",
)
LightSwarm routes work across specialized agents for role-based delegation and multi-agent task handling. Use it when one front-facing agent should delegate to domain agents such as support, HR, finance, research, or data analysis.
Keep each agent's role narrow, give each agent clear boundaries, and avoid letting delegated agents write unrelated long-term memory unless the memory policy allows it.
from LightAgent import LightAgent, LightSwarm
frontdesk = LightAgent(name="frontdesk", role="Route requests to the right specialist.")
finance = LightAgent(name="finance", role="Answer finance approval questions.")
swarm = LightSwarm()
swarm.register_agent(frontdesk, finance)
result = swarm.run(agent=frontdesk, query="How do I approve a vendor payment?")
For debugging role drift or cross-agent memory issues, see Multi-agent failure map.
agent.run(query, stream=True) returns OpenAI-compatible streaming chunks and remains backward compatible with existing integrations. Use it for chat UIs, long responses, and agent services that should start sending output before the full answer is complete.
For tool-heavy streaming tasks, max_tool_iterations can cap repeated tool-call loops independently from max_retry.
for chunk in agent.run("Write a short report about AI agents.", stream=True):
print(chunk)
Self-learning can be combined with a memory backend so an agent can retain scenario-specific knowledge across users or sessions. In production, pair self-learning with MemoryPolicy so low-quality, private, expired, or unrelated content does not enter long-term memory.
Recommended production controls:
namespace to separate tenants or environments.allowed_sources, allowed_scopes, and allowed_agent_names to limit retrieved memory.memory_write_admission, reject_write_patterns, min_write_length, and reject_duplicate_writes to control writes.enforce_expires_at=True when memory records carry expiration metadata.LightAgent can send trace data to Langfuse through tracetools configuration. For most local debugging, start with built-in trace events:
result = agent.run("Debug this workflow.", result_format="object", trace=True)
print(result.trace_id)
print(agent.export_trace())
See Trace Observability for event shapes and production debugging guidance.
LightEvaluator measures output, tool selection, policy events, recovery, latency, usage, and estimated cost against deterministic business cases.
LightFlow is the deterministic workflow layer. Use it when a task should run through known steps instead of relying on free-form agent planning.
Key v0.9.0 workflow controls:
pending, running, success, failed, skipped, waiting_approval.flow.validate(strict=True) checks unknown dependencies, cycles, and isolated steps.timeout, max_retry, cancel_if, fallback_agent, requires_approval, and approval_handler.JsonLightFlowStore stores run records and checkpoints.flow.resume(run_id) continues incomplete runs; flow.rerun_step(run_id, step_name) reruns one step and downstream steps.flow.get_run(run_id) and flow.list_runs() expose workflow state for UIs.from LightAgent import JsonLightFlowStore, LightAgent, LightFlow
store = JsonLightFlowStore(".lightflow_runs")
draft_agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
review_agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
backup_agent = LightAgent(model="gpt-4.1", api_key="your_api_key", base_url="your_base_url")
def approve(step, context):
return {"approved": True, "reason": "Approved by policy."}
flow = (
LightFlow(store=store)
.step("draft", agent=draft_agent, timeout=30, max_retry=2)
.step(
"review",
agent=review_agent,
depends_on=["draft"],
requires_approval=True,
approval_handler=approve,
fallback_agent=backup_agent,
)
)
validation = flow.validate(strict=True)
if validation["errors"]:
raise ValueError(validation["errors"])
result = flow.run("Prepare a customer support summary.", run_id="case-001", trace=True)
record = flow.get_run("case-001")
See LightFlow for complete workflow examples.
Guardrails are lightweight policy hooks around agent execution:
from LightAgent import (
LightAgent,
high_risk_parameter_guardrail,
output_redaction_guardrail,
privacy_input_guardrail,
sensitive_tool_confirmation_guardrail,
)
agent = LightAgent(
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
input_guardrails=[privacy_input_guardrail()],
tool_guardrails=[
sensitive_tool_confirmation_guardrail(["transfer_money"], approved=False),
high_risk_parameter_guardrail({
"amount": lambda value: value is not None and float(value) <= 1000,
}),
],
output_guardrails=[output_redaction_guardrail()],
)
Use default guardrail templates for privacy-sensitive input, sensitive tool confirmation, high-risk parameter validation, and output redaction. See Guardrails.
Runtime hooks are ordered middleware for production policies that need to observe, replace, or block lifecycle payloads without changing the default agent.run() behavior.
from LightAgent import HookDecision, LightAgent, PolicyHook
def redact_before_model(ctx):
if ctx.phase != "before_model_request":
return None
params = dict(ctx.payload["params"])
messages = list(params["messages"])
messages[-1] = {
**messages[-1],
"content": messages[-1]["content"].replace("secret-token", "[REDACTED]"),
}
params["messages"] = messages
return HookDecision.replace({"params": params})
def block_dangerous_tool(ctx):
if ctx.phase == "before_tool_call" and ctx.payload["tool_name"] == "delete_file":
return HookDecision.block("delete_file requires manual approval")
return None
agent = LightAgent(
model="gpt-4.1",
api_key="your_api_key",
base_url="your_base_url",
hooks=[
redact_before_model,
PolicyHook(block_dangerous_tool, phases={"before_tool_call"}, timeout=2.0),
],
)
result = agent.run("Summarize this secret-token safely.", result_format="object", trace=True)
print(result.content)
print(result.trace)
Hooks can target before_run, after_run, on_error, before_model_request, after_model_response, before_tool_call, after_tool_result, before_memory_retrieve, after_memory_retrieve, before_memory_write, after_memory_write, before_memory_promote, after_memory_promote, and on_handoff. Wrap security-sensitive callbacks with PolicyHook(...) to fail closed on exceptions or timeouts while ordinary observability hooks remain failure-isolated. LightFlow(hooks=[...]) also supports step lifecycle hooks such as before_flow_step, after_flow_step, on_approval_required, on_resume, and on_rerun. See Runtime Hooks and Runtime Hook Recipes.
SharedMemoryPool is an in-memory shared memory prototype for multi-agent experiments. It is append-first and keeps provenance metadata, making it useful for testing how multiple agents share information before adopting a durable vector or graph memory backend.
Use it with MemoryPolicy so each agent retrieves only memory that matches the expected namespace, source, scope, trust, confidence, or agent name.
Self-learning and delegation evidence become non-injectable memory candidates first; promote them with MemoryPolicy(memory_promotion_admission=...) or agent.promote_memory_candidate(candidate_id) before they can enter future prompts.
When upgrading from v0.9.4, audit legacy internal memory before backfilling promotion_status="promoted" and injectable=True; use require_promotion_for_internal_memory=False only as a temporary compatibility option. See Memory Admission.
LightAgent works with OpenAI-compatible chat completion endpoints, including OpenAI, OpenRouter, Zhipu ChatGLM, DeepSeek, Qwen, StepFun, Moonshot/Kimi, MiniMax, vLLM, llama.cpp, Ollama-compatible endpoints, and self-hosted gateways.
For provider-specific parameters, base URLs, local model setup, and troubleshooting, see Model Provider Configuration.
We welcome any form of contribution! Whether it's code, documentation, tests, or feedback, it's a tremendous help to the project. If you have great ideas or find bugs, please submit an Issue or Pull Request. Here are the contribution steps:
Fork button at the top right corner to copy the project to your GitHub repository.git checkout -b feature/YourFeature
git commit -m 'Add some feature'
git push origin feature/YourFeature
We will review your contributions promptly. Thank you for your support! ❤️
Shanghai Wanxing AI and Professor Zhang Liwen's research group from the School of Statistics and Data Science at Shanghai University of Finance and Economics have jointly open-sourced a new generation intelligent agent framework called LightAgent.The development and implementation of LightAgent owe much to the inspiration and support from the following open-source projects, especially the outstanding projects and teams:
LightAgent is licensed under the Apache 2.0 License. You can freely use, modify, and distribute this project, but please adhere to the terms of the license.
If you have any questions or suggestions, please feel free to contact Wanxing AI or Professor Zhang Liwen from the School of Statistics and Data Science at Shanghai University of Finance and Economics:
We look forward to your feedback and work together to make LightAgent even stronger! 🚀
LightAgent - Making intelligence lighter, making the future simpler. 🌈
LightAgent —— A lightweight, flexible, and powerful active Agent framework that assists you in quickly building intelligent applications!
@misc{2509.09292,
Author = {Weige Cai and Tong Zhu and Jinyi Niu and Ruiqi Hu and Lingyao Li and Tenglong Wang and Xiaowu Dai and Weining Shen and Liwen Zhang},
Title = {LightAgent: Production-level Open-source Agentic AI Framework},
Year = {2025},
Eprint = {arXiv:2509.09292},
Eprinttype = {arXiv},
Eprintclass = {cs.AI},
Url = {https://arxiv.org/abs/2509.09292},
Doi = {10.48550/arXiv.2509.09292},
Note = {Submitted on 11 Sep 2025}
}
Python
99.9%